_AsyncSQLAlchemyDep

setup

1
_AsyncSQLAlchemyDep.setup(engine: AsyncEngine)

Sets up the async session maker with the provided engine.

Args:

  • engine (AsyncEngine): The SQLAlchemy async engine to use for creating the session.

__call__

1
async _AsyncSQLAlchemyDep.__call__()

Returns an async SQLAlchemy session.

Raises:

  • ValueError: If the connection pool is not initialized.

_SQLAlchemyDep

__call__

1
_SQLAlchemyDep.__call__()

Returns a SQLAlchemy session.

Raises:

  • ValueError: If the connection pool is not initialized.

sqlalchemy_depends

1
sqlalchemy_depends(engine: Engine, autocommit: bool = False, autoflush: bool = False, expire_on_commit: bool = False)

Returns a dependency that returns a SQLAlchemy Session for use in FastAPI.

Args:

  • engine (Engine): A SQLAlchemy engine to use for creating the session.
  • autocommit (bool): Whether to autocommit the session after a commit.
  • autoflush (bool): Whether to autoflush the session after a commit.
  • expire_on_commit (bool): Whether to expire all instances after a commit.

Example usage:

1
import launchflow as lf
2
3
from fastapi import FastAPI, Depends
4
from sqlalchemy import text
5
from sqlalchemy.orm import Session
6
7
# Create the Postgres database instance
8
postgres = lf.gcp.CloudSQLPostgres("my-pg-db")
9
10
# Create the SQLAlchemy engine (connection pool)
11
engine = postgres.sqlalchemy_engine()
12
13
# Create the FastAPI dependency
14
session = lf.fastapi.sqlalchemy_depends(engine)
15
16
app = FastAPI()
17
18
@app.get("/")
19
def read_root(db: Session = Depends(session)):
20
# Use the session to query the database
21
return db.execute(text("SELECT 1")).scalar_one_or_none()

sqlalchemy_async_depends

1
sqlalchemy_async_depends(autocommit: bool = False, autoflush: bool = False, expire_on_commit: bool = False) -> _AsyncSQLAlchemyDep

Returns a dependency that returns a SQLAlchemy AsyncSession for use in FastAPI.

Args:

  • autocommit (bool): Whether to autocommit the session after a commit.
  • autoflush (bool): Whether to autoflush the session after a commit.
  • expire_on_commit (bool): Whether to expire all instances after a commit.

Example usage:

1
from contextlib import asynccontextmanager
2
3
import launchflow as lf
4
5
from fastapi import FastAPI, Depends
6
from sqlalchemy import text
7
from sqlalchemy.ext.asyncio import AsyncSession
8
9
# Create the Postgres database instance
10
postgres = lf.gcp.CloudSQLPostgres("my-pg-db")
11
12
# Create the FastAPI dependency
13
async_session = lf.fastapi.sqlalchemy_async_depends()
14
15
16
@asynccontextmanager
17
async def lifespan(app: FastAPI):
18
# Create the SQLAlchemy async engine (connection pool)
19
engine = await postgres.sqlalchemy_async_engine()
20
# Setup the async session dependency
21
async_session.setup(engine)
22
yield
23
24
app = FastAPI(lifespan=lifespan)
25
26
@app.get("/")
27
async def query_postgres(db: AsyncSession = Depends(async_session)):
28
# Use the async session to query the database
29
return await db.execute(text("SELECT 1")).scalar_one_or_none()