### Set up pgcrypto extension for PostgreSQL GUID server default Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/guid-type Call setup_guids_postgresql with a SQLAlchemy engine to install the pgcrypto extension, required for GUID_SERVER_DEFAULT_POSTGRESQL to work. ```python import sqlalchemy as sa from fastapi_utils.guid_type import setup_guids_postgresql database_uri = "postgresql://user:password@db:5432/app" engine = sa.create_engine(database_uri) setup_guids_postgresql(engine) ``` -------------------------------- ### Install fastapi-utils with pip Source: https://fastapiutils.github.io/fastapi-utils Install the basic slim package, add SQLAlchemy session maker, or install all packages. ```bash pip install fastapi-utils # For basic slim package :) pip install fastapi-utils[session] # To add sqlalchemy session maker pip install fastapi-utils[all] # For all the packages ``` -------------------------------- ### Set up development environment with make develop Source: https://fastapiutils.github.io/fastapi-utils/contributing Creates a virtual environment and installs dependencies using Poetry. Requires python3 and poetry to be installed. Prints the path to the executable at the end. ```bash $ make develop ``` -------------------------------- ### Complete example of using @cbv decorator with shared dependencies Source: https://fastapiutils.github.io/fastapi-utils/user-guide/class-based-views This example shows the full setup and usage of the @cbv decorator. It creates an APIRouter, defines a class ItemCBV decorated with @cbv(router), adds shared dependencies as class attributes (session and user_id), and uses them in endpoint methods via self.session and self.user_id. The example includes setup code for SQLAlchemy models, Pydantic models, and dependency functions. ```python from typing import NewType, Optional from uuid import UUID import sqlalchemy as sa from fastapi import APIRouter, Depends, FastAPI, Header, HTTPException from sqlalchemy.orm import Session, declarative_base from starlette.status import HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND from fastapi_utils.api_model import APIMessage, APIModel from fastapi_utils.cbv import cbv from fastapi_utils.guid_type import GUID # Begin Setup UserID = NewType("UserID", UUID) ItemID = NewType("ItemID", UUID) Base = declarative_base() class ItemORM(Base): __tablename__ = "item" item_id = sa.Column(GUID, primary_key=True) owner = sa.Column(GUID, nullable=False) name = sa.Column(sa.String, nullable=False) class ItemCreate(APIModel): name: str owner: UserID class ItemInDB(ItemCreate): item_id: ItemID def get_jwt_user(authorization: str = Header(...)) -> UserID: """Pretend this function gets a UserID from a JWT in the auth header""" def get_db() -> Session: """Pretend this function returns a SQLAlchemy ORM session""" def get_owned_item(session: Session, owner: UserID, item_id: ItemID) -> ItemORM: item: Optional[ItemORM] = session.get(ItemORM, item_id) if item is not None and item.owner != owner: raise HTTPException(status_code=HTTP_403_FORBIDDEN) if item is None: raise HTTPException(status_code=HTTP_404_NOT_FOUND) return item # End Setup app = FastAPI() router = APIRouter() # Step 1: Create a router @cbv(router) # Step 2: Create and decorate a class to hold the endpoints class ItemCBV: # Step 3: Add dependencies as class attributes session: Session = Depends(get_db) user_id: UserID = Depends(get_jwt_user) @router.post("/item") def create_item(self, item: ItemCreate) -> ItemInDB: # Step 4: Use `self.` to access shared dependencies item_orm = ItemORM(name=item.name, owner=self.user_id) self.session.add(item_orm) self.session.commit() return ItemInDB.from_orm(item_orm) @router.get("/item/{item_id}") def read_item(self, item_id: ItemID) -> ItemInDB: item_orm = get_owned_item(self.session, self.user_id, item_id) return ItemInDB.from_orm(item_orm) @router.put("/item/{item_id}") def update_item(self, item_id: ItemID, item: ItemCreate) -> ItemInDB: item_orm = get_owned_item(self.session, self.user_id, item_id) item_orm.name = item.name self.session.add(item_orm) self.session.commit() return ItemInDB.from_orm(item_orm) @router.delete("/item/{item_id}") def delete_item(self, item_id: ItemID) -> APIMessage: item = get_owned_item(self.session, self.user_id, item_id) self.session.delete(item) self.session.commit() return APIMessage(detail=f"Deleted item {item_id}") app.include_router(router) ``` -------------------------------- ### Add timing middleware and record timing example Source: https://fastapiutils.github.io/fastapi-utils/user-guide/timing-middleware Complete example showing how to add timing middleware to a FastAPI app, use record_timing for intermediate timing, and the expected logged output for various endpoints. The middleware logs timing records as single lines, with the example splitting them for readability. The exclude parameter prevents logging for routes containing 'untimed'. ```python import asyncio import logging from fastapi import FastAPI from starlette.requests import Request from starlette.staticfiles import StaticFiles from starlette.testclient import TestClient from fastapi_utils.timing import add_timing_middleware, record_timing logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI() add_timing_middleware(app, record=logger.info, prefix="app", exclude="untimed") static_files_app = StaticFiles(directory=".") app.mount(path="/static", app=static_files_app, name="static") @app.get("/timed") async def get_timed() -> None: await asyncio.sleep(0.05) @app.get("/untimed") async def get_untimed() -> None: await asyncio.sleep(0.1) @app.get("/timed-intermediate") async def get_with_intermediate_timing(request: Request) -> None: await asyncio.sleep(0.1) record_timing(request, note="halfway") await asyncio.sleep(0.1) TestClient(app).get("/timed") # INFO:__main__:TIMING: Wall: 53.0ms # | CPU: 1.2ms # | app.__main__.get_timed TestClient(app).get("/untimed") # TestClient(app).get("/timed-intermediate") # INFO:__main__:TIMING: Wall: 105.3ms # | CPU: 0.4ms # | app.__main__.get_with_intermediate_timing (halfway) # INFO:__main__:TIMING: Wall: 206.7ms # | CPU: 1.1ms # | app.__main__.get_with_intermediate_timing TestClient(app).get("/static/test") # INFO:__main__:TIMING: Wall: 1.6ms # | CPU: 1.6ms # | StaticFiles<'static'> ``` -------------------------------- ### FastAPISessionMaker usage with SQLAlchemy ORM table Source: https://fastapiutils.github.io/fastapi-utils/user-guide/session Complete example showing how to use FastAPISessionMaker with FastAPI. It defines a User ORM table, a DBSettings class that reads the database URI from the environment, a get_db dependency that yields a session, and a FastAPI route that queries the user by ID. Requires installation with `pip install fastapi-restful[session]` or `pip install fastapi-restful[all]`. ```python from functools import lru_cache from typing import Iterator from uuid import UUID import sqlalchemy as sa from fastapi import Depends, FastAPI from pydantic import BaseSettings from sqlalchemy.orm import Session, declarative_base from fastapi_utils.guid_type import GUID, GUID_DEFAULT_SQLITE from fastapi_utils.session import FastAPISessionMaker Base = declarative_base() class User(Base): __tablename__ = "user" id = sa.Column(GUID, primary_key=True, default=GUID_DEFAULT_SQLITE) name = sa.Column(sa.String, nullable=False) class DBSettings(BaseSettings): """Parses variables from environment on instantiation""" database_uri: str # could break up into scheme, username, password, host, db def get_db() -> Iterator[Session]: """FastAPI dependency that provides a sqlalchemy session""" yield from _get_fastapi_sessionmaker().get_db() @lru_cache() def _get_fastapi_sessionmaker() -> FastAPISessionMaker: """This function could be replaced with a global variable if preferred""" database_uri = DBSettings().database_uri return FastAPISessionMaker(database_uri) app = FastAPI() @app.get("/{user_id}") def get_user_name(db: Session = Depends(get_db), *, user_id: UUID) -> str: user = db.get(User, user_id) username = user.name return username ``` -------------------------------- ### Define a Resource class with get method Source: https://fastapiutils.github.io/fastapi-utils/user-guide/class-resource Create a class inheriting from Resource and define a get method that returns a string. This is the basic pattern for a resource endpoint. ```python from fastapi_utils import Resource class MyApi(Resource): def get(self): return "done" ``` -------------------------------- ### Add resource to FastAPI app in app.py Source: https://fastapiutils.github.io/fastapi-utils/user-guide/class-resource Create a FastAPI app, instantiate Api, add the resource with a URI, and return the app. This is the minimal setup to expose the resource. ```python from fastapi import FastAPI from docs.src.class_resource_view1 import MyApi from fastapi_utils import Api def create_app(): app = FastAPI() api = Api(app) myapi = MyApi() api.add_resource(myapi, "/uri") return app main = create_app() ``` -------------------------------- ### Set PYTHONPATH for editor tests Source: https://fastapiutils.github.io/fastapi-utils/contributing Example .env file for VS Code to add ./docs/src to PYTHONPATH for integrated tests. ```bash PYTHONPATH=./docs/src ``` -------------------------------- ### FastAPI dependency for SQLAlchemy session using FastAPISessionMaker Source: https://fastapiutils.github.io/fastapi-utils/user-guide/session Complete example showing how to set up a FastAPI app with a SQLAlchemy session dependency. It defines a User model with a GUID primary key, a DBSettings class that reads DATABASE_URI from the environment, and a get_db dependency that yields a session. The endpoint uses the session to fetch a user by UUID. Requires the DATABASE_URI environment variable to be set, otherwise a ValidationError is raised. ```python from functools import lru_cache from typing import Iterator from uuid import UUID import sqlalchemy as sa from fastapi import Depends, FastAPI from pydantic import BaseSettings from sqlalchemy.orm import Session, declarative_base from fastapi_utils.guid_type import GUID, GUID_DEFAULT_SQLITE from fastapi_utils.session import FastAPISessionMaker Base = declarative_base() class User(Base): __tablename__ = "user" id = sa.Column(GUID, primary_key=True, default=GUID_DEFAULT_SQLITE) name = sa.Column(sa.String, nullable=False) class DBSettings(BaseSettings): """Parses variables from environment on instantiation""" database_uri: str # could break up into scheme, username, password, host, db def get_db() -> Iterator[Session]: """FastAPI dependency that provides a sqlalchemy session""" yield from _get_fastapi_sessionmaker().get_db() @lru_cache() def _get_fastapi_sessionmaker() -> FastAPISessionMaker: """This function could be replaced with a global variable if preferred""" database_uri = DBSettings().database_uri return FastAPISessionMaker(database_uri) app = FastAPI() @app.get("/{user_id}") def get_user_name(db: Session = Depends(get_db), *, user_id: UUID) -> str: user = db.get(User, user_id) username = user.name return username ``` -------------------------------- ### Create a SQLAlchemy table with a GUID primary key Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/guid-type Use the declarative API to define a table with a GUID primary key. The GUID type is backend-agnostic and avoids issues with the postgres-specific UUID type. ```python import sqlalchemy as sa from sqlalchemy.orm import declarative_base from fastapi_utils.guid_type import GUID Base = declarative_base() class User(Base): __tablename__ = "user" id = sa.Column(GUID, primary_key=True) name = sa.Column(sa.String, nullable=False) related_id = sa.Column(GUID) # a nullable, related field ``` -------------------------------- ### Request body in camelCase Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/api-model Example JSON request body using camelCase field names. Both snake_case and camelCase are accepted by APIModel. ```json { "userId": "00000000-0000-0000-0000-000000000000", "emailAddress": "user@email.com" } ``` -------------------------------- ### Request body in snake_case Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/api-model Example JSON request body using snake_case field names. Both snake_case and camelCase are accepted by APIModel. ```json { "user_id": "00000000-0000-0000-0000-000000000000", "email_address": "user@email.com" } ``` -------------------------------- ### Add a PostgreSQL server default for GUID primary key Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/guid-type Use GUID_SERVER_DEFAULT_POSTGRESQL to set a server-side default of "gen_random_uuid()". This requires the pgcrypto extension in PostgreSQL. ```python import sqlalchemy as sa from sqlalchemy.orm import declarative_base from fastapi_utils.guid_type import GUID, GUID_SERVER_DEFAULT_POSTGRESQL Base = declarative_base() class User(Base): __tablename__ = "user" id = sa.Column( GUID, primary_key=True, server_default=GUID_SERVER_DEFAULT_POSTGRESQL ) name = sa.Column(sa.String, nullable=False) related_id = sa.Column(GUID) ``` -------------------------------- ### @repeat_every decorator Source: https://fastapiutils.github.io/fastapi-utils/user-guide/repeated-tasks Decorator that starts a loop and calls the wrapped function periodically with a delay specified by the seconds argument. It can be combined with FastAPI startup events to run tasks repeatedly while the server is running. ```APIDOC ## @repeat_every decorator ### Description When a function decorated with the `@repeat_every(...)` decorator is called, a loop is started, and the function is called periodically with a delay determined by the `seconds` argument provided to the decorator. If you also apply the `@app.event("startup")` decorator, FastAPI will call the function during server startup, and the function will then be called repeatedly while the server is still running. ### Usage ```python from fastapi import FastAPI from sqlalchemy.orm import Session from fastapi_utils.session import FastAPISessionMaker from fastapi_utils.tasks import repeat_every database_uri = f"sqlite:///./test.db?check_same_thread=False" sessionmaker = FastAPISessionMaker(database_uri) app = FastAPI() def remove_expired_tokens(db: Session) -> None: """Pretend this function deletes expired tokens from the database""" @app.on_event("startup") @repeat_every(seconds=60 * 60) # 1 hour def remove_expired_tokens_task() -> None: with sessionmaker.context_session() as db: remove_expired_tokens(db=db) ``` ### Parameters #### Keyword Arguments - **seconds** (float) - Required - The number of seconds to wait between successive calls. - **wait_first** (bool) - Optional - If `False` (the default), the wrapped function is called immediately when the decorated function is first called. If `True`, the decorated function will wait one period before making the first call to the wrapped function. - **logger** (Optional[logging.Logger]) - Optional - If you pass a logger, any exceptions raised in the repeating execution loop will be logged (with a traceback) to the provided logger. - **raise_exceptions** (bool) - Optional - If `False` (the default), exceptions are caught in the repeating execution loop, but are not raised. If `True`, an exception will be raised. Note that if an exception is raised, the repeated execution will stop. - **max_repetitions** (Optional[int]) - Optional - If `None` (the default), the decorated function will keep repeating forever. Otherwise, it will stop repeated execution after the specified number of calls. ### Notes - The wrapped function should not take any required arguments. - `repeat_every` works with both `async def` and `def` functions. - `repeat_every` is safe to use with `def` functions that perform blocking IO – they are executed in a threadpool (just like `def` endpoints). ``` -------------------------------- ### record_timing Source: https://fastapiutils.github.io/fastapi-utils/user-guide/timing-middleware Emits performance information for a request at the point it is called. Must be used with a FastAPI app that has the timing middleware installed via add_timing_middleware. ```APIDOC ## record_timing ### Description Called on a `starlette.requests.Request` instance for a FastAPI app with the timing middleware installed (via `add_timing_middleware`). Emits performance information for the request at the point at which it is called, allowing intermediate timing measurements within a request handler. ### Function Signature `record_timing(request: starlette.requests.Request, note: str = "")` ### Parameters - **request** (starlette.requests.Request) - Required - The request instance for which to record timing. - **note** (str) - Optional - A note to append to the route name in the timing output. Defaults to `""`. ### Example ```python from fastapi import FastAPI, Request from fastapi_utils.timing import add_timing_middleware, record_timing import asyncio app = FastAPI() add_timing_middleware(app) @app.get("/timed-intermediate") async def get_with_intermediate_timing(request: Request) -> None: await asyncio.sleep(0.1) record_timing(request, note="halfway") await asyncio.sleep(0.1) ``` ### Output Format When called, it emits a timing record similar to the middleware's output, with the note appended to the route name in parentheses. For example: ``` TIMING: Wall: 105.3ms | CPU: 0.4ms | app.__main__.get_with_intermediate_timing (halfway) ``` ``` -------------------------------- ### Create a model with APIModel Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/api-model Use APIModel instead of pydantic.BaseModel to enable snake_case attributes, camelCase JSON input, and orm_mode. The example defines a User model, a UserORM dataclass to simulate a SQLAlchemy model, and a FastAPI endpoint that returns a UserORM object. ```python from dataclasses import dataclass from typing import NewType from uuid import UUID from fastapi import FastAPI from fastapi_utils.api_model import APIModel UserID = NewType("UserID", UUID) class User(APIModel): user_id: UserID email_address: str @dataclass class UserORM: """ You can pretend this class is a SQLAlchemy model """ user_id: UserID email_address: str app = FastAPI() @app.post("/users", response_model=User) async def create_user(user: User) -> UserORM: return UserORM(user.user_id, user.email_address) ``` -------------------------------- ### Basic CRUD app with repeated dependencies Source: https://fastapiutils.github.io/fastapi-utils/user-guide/class-based-views A complete FastAPI CRUD application for items, showing create, read, update, and delete endpoints. Each endpoint repeats the same dependencies (get_db and get_jwt_user) via Depends, which the @cbv decorator aims to reduce. Includes setup for SQLAlchemy models, JWT user extraction, and database session handling. ```python from typing import NewType, Optional from uuid import UUID import sqlalchemy as sa from fastapi import Depends, FastAPI, Header, HTTPException from sqlalchemy.orm import Session, declarative_base from starlette.status import HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND from fastapi_utils.api_model import APIMessage, APIModel from fastapi_utils.guid_type import GUID # Begin setup UserID = NewType("UserID", UUID) ItemID = NewType("ItemID", UUID) Base = declarative_base() class ItemORM(Base): __tablename__ = "item" item_id = sa.Column(GUID, primary_key=True) owner = sa.Column(GUID, nullable=False) name = sa.Column(sa.String, nullable=False) class ItemCreate(APIModel): name: str class ItemInDB(ItemCreate): item_id: ItemID owner: UserID def get_jwt_user(authorization: str = Header(...)) -> UserID: """Pretend this function gets a UserID from a JWT in the auth header""" def get_db() -> Session: """Pretend this function returns a SQLAlchemy ORM session""" def get_owned_item(session: Session, owner: UserID, item_id: ItemID) -> ItemORM: item: Optional[ItemORM] = session.get(ItemORM, item_id) if item is not None and item.owner != owner: raise HTTPException(status_code=HTTP_403_FORBIDDEN) if item is None: raise HTTPException(status_code=HTTP_404_NOT_FOUND) return item # End setup app = FastAPI() @app.post("/item", response_model=ItemInDB) def create_item( *, session: Session = Depends(get_db), user_id: UserID = Depends(get_jwt_user), item: ItemCreate, ) -> ItemInDB: item_orm = ItemORM(name=item.name, owner=user_id) session.add(item_orm) session.commit() return ItemInDB.from_orm(item_orm) @app.get("/item/{item_id}", response_model=ItemInDB) def read_item( *, session: Session = Depends(get_db), user_id: UserID = Depends(get_jwt_user), item_id: ItemID, ) -> ItemInDB: item_orm = get_owned_item(session, user_id, item_id) return ItemInDB.from_orm(item_orm) @app.put("/item/{item_id}", response_model=ItemInDB) def update_item( *, session: Session = Depends(get_db), user_id: UserID = Depends(get_jwt_user), item_id: ItemID, item: ItemCreate, ) -> ItemInDB: item_orm = get_owned_item(session, user_id, item_id) item_orm.name = item.name session.add(item_orm) session.commit() return ItemInDB.from_orm(item_orm) @app.delete("/item/{item_id}", response_model=APIMessage) def delete_item( *, session: Session = Depends(get_db), user_id: UserID = Depends(get_jwt_user), item_id: ItemID, ) -> APIMessage: item = get_owned_item(session, user_id, item_id) session.delete(item) session.commit() return APIMessage(detail=f"Deleted item {item_id}") ``` -------------------------------- ### Use an application-side default for GUID primary key Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/guid-type Use GUID_DEFAULT_SQLITE as a default for the primary key column. This is an alias for uuid.uuid4 and works without a server default. ```python import sqlalchemy as sa from sqlalchemy.orm import declarative_base from fastapi_utils.guid_type import GUID, GUID_DEFAULT_SQLITE Base = declarative_base() class User(Base): __tablename__ = "user" id = sa.Column(GUID, primary_key=True, default=GUID_DEFAULT_SQLITE) name = sa.Column(sa.String, nullable=False) related_id = sa.Column(GUID) ``` -------------------------------- ### Use camel2snake to generate snake_case table names in SQLAlchemy Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/camelcase Apply camel2snake in a SQLAlchemy declarative base to automatically name tables in snake_case based on the class name. For example, a class MyUser would result in a table named my_user. ```python from sqlalchemy.orm import declarative_base, declared_attr from fastapi_utils.camelcase import camel2snake class CustomBase: @declared_attr def __tablename__(cls) -> str: return camel2snake(cls.__name__) Base = declarative_base(cls=CustomBase) ``` -------------------------------- ### Renaming enum members without updating values causes pydantic errors Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/enums If you rename enum members (e.g., via IDE refactoring) but forget to update the associated values, pydantic will only parse the original values. This example shows the resulting mismatch. ```python from enum import Enum class MyEnum(str, Enum): choice_a = "value_a" # pydantic would only parse "value_a" to MyEnum.choice_a choice_b = "value_b" ``` -------------------------------- ### Run live documentation server with bash scripts/docs-live.sh Source: https://fastapiutils.github.io/fastapi-utils/contributing Builds the site and live-reloads on changes. Serves documentation on http://0.0.0.0:8008. ```bash $ bash scripts/docs-live.sh ``` -------------------------------- ### List makefile recipes with make Source: https://fastapiutils.github.io/fastapi-utils/contributing Displays basic documentation of available makefile recipes. ```bash $ make ``` -------------------------------- ### Run all tests with make test Source: https://fastapiutils.github.io/fastapi-utils/contributing Runs the full test suite. ```bash $ make test ``` -------------------------------- ### Run linting with make lint Source: https://fastapiutils.github.io/fastapi-utils/contributing Runs flake8 for linting. ```bash $ make lint ``` -------------------------------- ### Auto-format code with make format Source: https://fastapiutils.github.io/fastapi-utils/contributing Runs black, autoflake8, and isort to format code, sort imports, and remove unused imports. ```bash $ make format ``` -------------------------------- ### Resource with dependencies via constructor Source: https://fastapiutils.github.io/fastapi-utils/user-guide/class-resource Pass a MongoClient instance to the resource constructor before adding it to the API. Initialization happens before adding the resource, so dependencies can be injected this way. ```python from fastapi import FastAPI from pymongo import MongoClient from docs.src.class_resource_view1 import MyApi from fastapi_utils import Api def create_app(): app = FastAPI() api = Api(app) mongo_client = MongoClient("mongodb://localhost:27017") myapi = MyApi(mongo_client) api.add_resource(myapi, "/uri") return app main = create_app() ``` -------------------------------- ### Activate virtual environment with poetry shell Source: https://fastapiutils.github.io/fastapi-utils/contributing Activates the Poetry-managed virtual environment. After activation, verify with 'which python' to ensure the binary path contains 'fastapi-utils'. ```bash $ poetry shell ``` -------------------------------- ### Schedule cleanup task with @repeat_every and @app.on_event("startup") Source: https://fastapiutils.github.io/fastapi-utils/user-guide/repeated-tasks Uses @repeat_every and @app.on_event("startup") to run a cleanup task hourly. Requires FastAPISessionMaker from fastapi_utils.session, and the wrapped function must not take required arguments. Works with async def and def; blocking IO in def functions runs in a threadpool. ```python from fastapi import FastAPI from sqlalchemy.orm import Session from fastapi_utils.session import FastAPISessionMaker from fastapi_utils.tasks import repeat_every database_uri = f"sqlite:///./test.db?check_same_thread=False" sessionmaker = FastAPISessionMaker(database_uri) app = FastAPI() def remove_expired_tokens(db: Session) -> None: """Pretend this function deletes expired tokens from the database""" @app.on_event("startup") @repeat_every(seconds=60 * 60) # 1 hour def remove_expired_tokens_task() -> None: with sessionmaker.context_session() as db: remove_expired_tokens(db=db) ``` -------------------------------- ### Configure FastAPI with get_api_settings and cache_clear Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/api-settings Use this pattern to create a FastAPI instance inside a function, clearing the lru_cache and reloading environment settings each time. The settings are read from environment variables like API_DEBUG and API_DISABLE_DOCS, and passed as keyword arguments via settings.fastapi_kwargs. ```python from fastapi import FastAPI from fastapi_utils.api_settings import get_api_settings def get_app() -> FastAPI: get_api_settings.cache_clear() settings = get_api_settings() app = FastAPI(**settings.fastapi_kwargs) # return app ``` -------------------------------- ### Verify virtual environment with which python Source: https://fastapiutils.github.io/fastapi-utils/contributing Checks that the active Python binary is from the virtual environment. The output should show a path containing 'fastapi-utils'. ```bash $ which python some/directory/fastapi-utils-SOMETHING-py3.X/bin/python ``` -------------------------------- ### Generate coverage report with make testcov Source: https://fastapiutils.github.io/fastapi-utils/contributing Generates a coverage report. On MacOS, opens the report in a browser; on other OSes, prints a link to the local HTML. ```bash make testcov ``` -------------------------------- ### Run static type checking with make mypy Source: https://fastapiutils.github.io/fastapi-utils/contributing Runs mypy for static type checking. ```bash $ make mypy ``` -------------------------------- ### Using fastapi_utils.enums.StrEnum to auto-generate string values Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/enums Inherit from StrEnum to automatically generate string values that match the member names, reducing boilerplate. The assert statements show that name and value are equal. ```python from enum import auto from fastapi_utils.enums import StrEnum class MyEnum(StrEnum): choice_a = auto() choice_b = auto() assert MyEnum.choice_a.name == MyEnum.choice_a.value == "choice_a" assert MyEnum.choice_b.name == MyEnum.choice_b.value == "choice_b" ``` -------------------------------- ### add_timing_middleware Source: https://fastapiutils.github.io/fastapi-utils/user-guide/timing-middleware Adds a timing middleware to a FastAPI app that logs basic profiling information for each request. Accepts optional parameters for custom logging, prefixing route names, and excluding routes. ```APIDOC ## add_timing_middleware ### Description Adds a timing middleware to a FastAPI app that logs basic profiling information (wall time and CPU time) for each request. The middleware is designed to have low overhead and be suitable for production use. ### Function Signature `add_timing_middleware(app: FastAPI, record: Optional[Callable[[str], None]] = None, prefix: str = "", exclude: Optional[str] = None)` ### Parameters - **app** (FastAPI) - Required - The FastAPI app to which to add the timing middleware. - **record** (Optional[Callable[[str], None]]) - Optional - The callable to call on the generated timing messages. Defaults to `print`. A good choice is the `info` method of a `logging.Logger` instance. - **prefix** (str) - Optional - A prefix to prepend to the generated route names. Useful for distinguishing between mounted ASGI apps. Defaults to `""`. - **exclude** (Optional[str]) - Optional - If provided, any route whose generated name includes this value will not have its timing stats recorded. ### Example ```python from fastapi import FastAPI from fastapi_utils.timing import add_timing_middleware import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI() add_timing_middleware(app, record=logger.info, prefix="app", exclude="untimed") ``` ### Output Format Each timing record is logged as a single line with the following structure: ``` TIMING: Wall: ms | CPU: ms | ``` Where `` is the generated route name (e.g., `app.__main__.get_timed`). ``` -------------------------------- ### Basic string-valued enum with str and Enum Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/enums Create a string-valued enum by inheriting from both str and enum.Enum. This ensures proper OpenAPI encoding when used with pydantic/FastAPI. ```python from enum import Enum class MyEnum(str, Enum): value_a = "value_a" value_b = "value_b" ``` -------------------------------- ### Convert between camelCase and snake_case with camel2snake and snake2camel Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/camelcase Use these functions to convert strings between camelCase and snake_case. snake2camel accepts a start_lower flag to control the case of the first letter. ```python from fastapi_utils.camelcase import camel2snake, snake2camel assert snake2camel("some_field_name", start_lower=False) == "SomeFieldName" assert snake2camel("some_field_name", start_lower=True) == "someFieldName" assert camel2snake("someFieldName") == "some_field_name" assert camel2snake("SomeFieldName") == "some_field_name" ``` -------------------------------- ### Resource with @set_responses decorator Source: https://fastapiutils.github.io/fastapi-utils/user-guide/class-resource Declare response models and status codes for resource methods using @set_responses. Supports multiple status codes and custom response models, and also accepts kwargs of the original function. ```python from pydantic import BaseModel from fastapi_utils import Resource, set_responses # Setup class ResponseModel(BaseModel): answer: str class ResourceAlreadyExistsModel(BaseModel): is_found: bool class ResourceModel(BaseModel): ID: str name: str # Setup end class MyApi(Resource): def __init__(self, mongo_client): self.mongo = mongo_client @set_responses(ResponseModel) def get(self): return "Done" @set_responses(ResponseModel, 200) def put(self): return "Redone" @set_responses( ResponseModel, 201, { 409: { "description": "The path can't be found", "model": ResourceAlreadyExistsModel, } }, ) def post(self, res: ResourceModel): if self.mongo.is_resource_exist(res.name): return JSONResponse(409, content={"is_found": true}) return "Done again" ``` -------------------------------- ### Using fastapi_utils.enums.CamelStrEnum for camelCase values Source: https://fastapiutils.github.io/fastapi-utils/user-guide/basics/enums Inherit from CamelStrEnum to automatically generate camelCase string values from the member names. The assert statements show the generated values. ```python from enum import auto from fastapi_utils.enums import CamelStrEnum class MyEnum(CamelStrEnum): choice_a = auto() choice_b = auto() assert MyEnum.choice_a.name == MyEnum.choice_a.value == "choiceOne" assert MyEnum.choice_b.name == MyEnum.choice_b.value == "choiceTwo" ``` -------------------------------- ### Simplify operation IDs with simplify_operation_ids Source: https://fastapiutils.github.io/fastapi-utils/user-guide/openapi Uses fastapi_utils.openapi.simplify_operation_ids to replace the generated operationId with one based only on the function name. Requires unique function names per endpoint/method combination to avoid conflicts. ```python from fastapi import FastAPI from fastapi_utils.openapi import simplify_operation_ids app = FastAPI() @app.get("/api/v1/resource/{resource_id}") def get_resource(resource_id: int) -> int: return resource_id simplify_operation_ids(app) path_spec = app.openapi()["paths"]["/api/v1/resource/{resource_id}"] operation_id = path_spec["get"]["operationId"] assert operation_id == "get_resource" ``` -------------------------------- ### Default FastAPI operationId generation Source: https://fastapiutils.github.io/fastapi-utils/user-guide/openapi Shows the default operationId generated by FastAPI, which includes the function name, endpoint path, and request method. The assertion verifies the exact operationId string. ```python from fastapi import FastAPI app = FastAPI() @app.get("/api/v1/resource/{resource_id}") def get_resource(resource_id: int) -> int: return resource_id path_spec = app.openapi()["paths"]["/api/v1/resource/{resource_id}"] operation_id = path_spec["get"]["operationId"] assert operation_id == "get_resource_api_v1_resource__resource_id__get" ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.