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
1import launchflow as lf
2
3gcs_bucket = lf.gcp.GCSBucket('my-bucket')
main.py
1from app.infra import gcs_bucket
2from fastapi import FastAPI, Depends
3from google.cloud import storage
4
5@asynccontextmanager
6async def lifespan(app: FastAPI):
7 await gcs_bucket.connect_async()
8 yield
9
10app = FastAPI(lifespan=lifespan)
11
12@app.get("/")
13async 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
1import launchflow as lf
2
3db = lf.gcp.CloudSQLPostgres("my-pg-db")
main.py
1from app.infra import db
2from app.models import Base
3from fastapi import FastAPI, Depends
4from sqlalchemy.orm import Session
5
6db.connect()
7engine = db.sqlalchemy_engine()
8Base.metadata.create_all(bind=engine)
9get_db = launchflow.fastapi.sqlalchemy(engine)
10
11app = FastAPI()
12
13@app.get("/")
14def read_root(db: Session = Depends(get_db)):
15 return db.execute(text("SELECT 1")).scalar_one_or_none()