launchflow.fastapi

1
def sqlalchemy_depends(engine: Engine,
2
                       autocommit: bool = False,
3
                       autoflush: bool = False,
4
                       expire_on_commit: bool = False)
content_copy

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()
content_copy

1
def sqlalchemy_async_depends(autocommit: bool = False,
2
                             autoflush: bool = False,
3
                             expire_on_commit: bool = False)
content_copy

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()
content_copy