Framework Guides
FastAPI
Overview
All of LaunchFlow's resource can be easily plugged in and used in your FastAPI application using FastAPI Dependency Injection. 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
5app = FastAPI(lifespan=lifespan)
6
7@app.get("/")
8async def read_root(bucket: storage.Bucket = Depends(gcs_bucket.bucket)) -> bytes:
9 return bucket.blob("my-file").download_as_bytes()
You can also make use of FastAPI's lifespan events to fetch the connection info of your resources when your application starts up. After it's been fetched once, it'll be cached for subsequent uses:
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.outputs_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 allows you 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.outputs()
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()