FastAPI Utilities

https://fastapiutils.github.io/fastapi-utils/
Admin
FastAPI Utilities provides a suite of reusable tools and helpers designed to reduce boilerplate and...
Tokens:15,240
Snippets:142
Trust Score:9.7
Update:6 days ago
Show doc for...
Tokens:
Raw
### 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.<dependency_name>` 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")
# <nothing logged>

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)
    # <Typically, you would include endpoint routers here>
    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:   <wall_time>ms
  | CPU:    <cpu_time>ms
  | <route_name>
```
Where `<route_name>` 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"

```
Tokens:
Raw