-
Notifications
You must be signed in to change notification settings - Fork 0
FastAPI Programming Notes
In FastAPI endpoints you need a reference to a Session and/or an Engine.
These are usually injected using FastAPI's Depends, which requires you supply
a generator for database sessions. In my code, the generator is defined in the Database class.
In general the code is something like (this isn't how I did it):
DATABASE_URL = "sqlite+aiosqlite:///./example.sqlite3"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": True}) # False?
# a factory method for Sessions
SessionMaker = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_session() -> Session:
session = SessionMaker()
try:
yield session
finally:
session.close()Dependency injection using Depends:
router = APIRouter(tags=["users"])
@router.post("/users/", response_model=schemas.User)
def create_user(user: User, session: Session = Depends(get_session)) -> schemas.User:
session.add(**user)
session.commit()
session.refresh(user)FastAPI has a lot of (sometimes cludgy) optional parameters that you can access in route handlers (functions). These include the HTTP Request and Response objects, query parameters, and specific request headers.
To get the Request and Response objects in a route handler, add parameters with type hints:
from fastapi import APIRouter, Depends, Request, Response
@router.post("/users", ...)
async def create_user(user_data, request: Request, response: Response, ...):A POST request that creates a new resource (e.g. entity) should return the URL of the new resource
using the HTTP Location header. Location can be a relative URL, that is, not including the protocol or host name.
For example:
from fastapi import Depends, Request, Response
@router.post("/users", ...)
async def create_user(user_data: schemas.UserCreate,
request: Request, response: Response,
session = Depends(get_session)):
# Validate the request data and then...
result: models.User = await user_dao.create(session, user_data)
# Add Location of the new user. "url_for" performs reverse mapping
user_id = result.id
location = request.url_for("get_user", user_id=str(user_id))
response.headers["Location"] = str(location)
# result is serialized by FastAPI and used as response body
return resultIn the FastAPI @router decorator you can specify a response_model and on the function header you can add a type hint for the return value. For example:
@router.get("/users/{user_id}", response_model=schemas.User)
async def get_user(user_id: int, session=Depends(get_session)) -> schemas.User:
...Seems redundant.
The response_model and return type hint serve complimentary purposes:
- Primarily for editor support and type checking.
- FastAPI does not use it to generate OpenAPI documentation or enforce response serialization.
- The return type is the type of the response body, not
fastapi.Response. Unlike Django.
This FastAPI directive controls:
- Serialization: filters response fields according to the schema.
- Validation: ensures the returned data matches the Pydantic schema.
-
OpenAPI documentation: populates the response schema in the generated API docs (
/docsor/openapi.json).
If a validation error occurs it probably means a programming error or deployment error, i.e. database structure doesn't match code for schema.
Specify both.
In my router functions I return an ORM model object (models.User) not a Pydantic schema object. FastAPI converts this to a schema object itself. Then it serializes the return value into the body of the Response.