SQLAlchemy minimal setup for SQLite
A minimal example for managing an SQLite database with Python via SQLAlchemy library. Includes a definition of a relationship between tables for the purpose of implementing a dictionary.
Conda Installation
conda activate YourCondaEnvironment
conda install -c anaconda sqlalchemy
Directory Structure
/scripts_directory
├─ my_database.db
├─ database.py
├─ models.py
├─ crud.py
└─ create_tables.py
Source Code
database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from pathlib import Path
# SQLite database in the same folder as python file
SQLITE_PATH = "sqlite:///" + str( Path(__file__).parents[0].resolve().as_posix()) + "/my_database.db"
engine = create_engine(SQLITE_PATH, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Full path to SQLite database file is automatically inferred from the folder the script itself is in.
models.py
from sqlalchemy import Column, ForeignKey, Integer, String, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from database import Base
class TableA(Base):
"""Example dictionary table"""
__tablename__ = "table_a"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, nullable=True)
single_object_from_table_a = relationship("TableB", back_populates="multiple_objects_from_table_b")
class TableB(Base):
"""Example regular table"""
__tablename__ = "table_b"
id = Column(Integer, primary_key=True, index=True)
table_a_id = Column(Integer, ForeignKey("table_a.id"), nullable=True)
row_created = Column(DateTime, default=func.now(), nullable=True)
multiple_objects_from_table_b = relationship("TableA", back_populates="single_object_from_table_a")
crud.py
from models import *
from database import SessionLocal, engine
def table_b_add_new_row(dict_table_id:int):
"""Add a new row to Table B"""
with SessionLocal() as db:
new_row = TableB(table_a_id=dict_table_id)
db.add(new_row)
db.commit()
def table_b_row_exists(row_id:int):
"""Check if Table B row with given id exists"""
with SessionLocal() as db:
if db.query(TableB).filter(TableB.id == row_id).first() is not None:
return True
else:
return False
def table_a_row_exists(row_id:int):
"""Check if Table A row with given id exists"""
with SessionLocal() as db:
if db.query(TableA).filter(TableA.id == row_id).first() is not None:
return True
else:
return False
def table_b_update_dict_column(row_id:int, dict_table_id:int):
"""Update the dictionary attribute field for Table B row with given id"""
if table_b_row_exists(row_id=row_id) and table_a_row_exists(row_id=dict_table_id):
with SessionLocal() as db:
db.query( TableB ).filter( TableB.id == row_id ).update( {'table_a_id': dict_table_id}, synchronize_session=False )
db.commit()
else:
print("Couldn't find given id in tables!")
create_tables.py
from models import *
from database import engine, SessionLocal, Base
def create_tables():
Base.metadata.create_all(bind=engine)
def fill_dicts():
"""Fill the dictionary tables"""
with SessionLocal() as db:
db.add(TableA(id=0,name="VALUE A"))
db.add(TableA(id=1,name="VALUE B"))
db.commit()
if __name__ == '__main__':
create_tables()
fill_dicts()
Running create_tables.py automaticall creates the database file under the path defined in database.py, creates all the tables defines in models.py, and fills the dictionary table with predefined values.