Framework Guides

FastAPI

Overview

All of LaunchFlow's resource can be easily plugged in and used in your FastAPI application using FastAPI Dependency Injection, and FastAPI lifespan events.

For all resources you can call .connect() in the lifespan event to connect to them in an async manner.

Most resource clients can simply be injected as is. For example if you are using a GCS or S3 bucket you can do the following:

infra.py

1
import launchflow as lf
2
3
gcs_bucket = lf.gcp.GCSBucket('my-bucket')

main.py

1
from app.infra import gcs_bucket
2
from fastapi import FastAPI, Depends
3
from google.cloud import storage
4
5
@asynccontextmanager
6
async def lifespan(app: FastAPI):
7
await gcs_bucket.connect_async()
8
yield
9
10
app = FastAPI(lifespan=lifespan)
11
12
@app.get("/")
13
async def read_root(bucket: storage.Bucket = Depends(gcs_bucket.bucket)) -> bytes:
14
return bucket.blob("my-file").download_as_bytes()

Complex Dependencies

LaunchFlow includes more complex dependencies in the launchflow.fastapi package. This can allow to do things such as create a SQLAlchemy Session per request to your fast API application.

SQLAlchemy

infra.py

1
import launchflow as lf
2
3
db = lf.gcp.CloudSQLPostgres("my-pg-db")

main.py

1
from app.infra import db
2
from app.models import Base
3
from fastapi import FastAPI, Depends
4
from sqlalchemy.orm import Session
5
6
db.connect()
7
engine = db.sqlalchemy_engine()
8
Base.metadata.create_all(bind=engine)
9
get_db = launchflow.fastapi.sqlalchemy(engine)
10
11
app = FastAPI()
12
13
@app.get("/")
14
def read_root(db: Session = Depends(get_db)):
15
return db.execute(text("SELECT 1")).scalar_one_or_none()
Previous
Use LaunchFlow with an Existing Application