Skip to content

Database

create_db_engine(db_url=DB_URL)

Create a SQLAlchemy engine.

Source code in germinate_ai/data/database.py 
12
13
14
15
def create_db_engine(db_url=DB_URL):
    """Create a SQLAlchemy engine."""
    engine = create_engine(db_url)
    return engine

create_tables(engine=None)

Create all defined tables -- DEV ONLY.

Source code in germinate_ai/data/database.py 
30
31
32
33
34
35
def create_tables(engine=None):
    """Create all defined tables -- DEV ONLY."""
    if engine is None:
        engine = create_db_engine(db_url=DB_URL)
    logger.info("Creating all tables...")
    Base.metadata.create_all(bind=engine)

get_db()

Get a SQLALchemy session.

Source code in germinate_ai/data/database.py 
38
39
40
41
42
def get_db():
    """Get a SQLALchemy session."""
    Session = get_db_session()
    with Session() as db:
        yield db

get_db_session(engine=None, db_url=DB_URL)

Create a SQLAlchemy session factory.

Source code in germinate_ai/data/database.py 
18
19
20
21
22
23
def get_db_session(engine=None, db_url: str = DB_URL):
    """Create a SQLAlchemy session factory."""
    if engine is None:
        engine = create_db_engine(db_url)
    Session = sessionmaker(autoflush=False, bind=engine)
    return Session

test_db_connection(db)

Test connection to DB.

Source code in germinate_ai/data/database.py 
45
46
47
48
49
50
51
def test_db_connection(db: Session) -> Exception | Literal[True]:
    """Test connection to DB."""
    try:
        db.execute(select(1))
        return True
    except Exception as e:
        return e