Pydantic / FastAPI quick notes

Personal quick notes to work with FastAPI / Pydantic.

Pydantic acceptinc multiple payloads with discrimination field
Default values, documentation, description and examples using Field
Simple but powerful row level action authorization

This post is a published unpolished draft notes.

Pydantic multiple payloads

Use an annotated union with a field descriminator to accept multiple payloads or variations of the same payload.

from pydantic import BaseModel, Field, computed_field, ConfigDict
from typing import List, Annotated, Union, Literal, Any, Optional

import enum
import uuid
from datetime import datetime, timedelta
from typing import List, Annotated, Union, Literal, Any, Optional
from uuid import UUID

from pydantic import BaseModel, Field, computed_field, ConfigDict

class QuoteItemBase(BaseModel):
    type: Literal["item"]
    description: str | None = None
    special_requirements: str | None = None

class QuoteItemTranslationBase(QuoteItemBase):
    type: Literal["translation"]
    source_language: str
    target_language: str

class QuoteItemEmailDeliveryBase(QuoteItemBase):
    type: Literal["delivery_email"]
    description: str | None = "Email delivery"
    address: str
    name: str

class QuoteItemPostDeliveryBase(QuoteItemBase):
    type: Literal["delivery_post"]
    description: str | None = "Post delivery"
    name: str
    address1: str
    address2: str
    address3: str
    post_code: str
    city: str
    country: str
    copies: int = 1

    delivery_class: str | None = None
    weight: int | None = 0

class QuoteItemVoucherBase(QuoteItemBase):
    type: Literal["voucher"]
    code: str

AnyQuoteItemBase = Annotated[Union[
    QuoteItemBase,
    QuoteItemTranslationBase,
    QuoteItemEmailDeliveryBase,
    QuoteItemPostDeliveryBase,
    QuoteItemVoucherBase
], Field(discriminator="type")]

@router.post("/{quote_id}")
def add_quote_item(
    quote_id: UUID,
    quote: model.Quote = Permission("view", QuoteRepository.get),
    item: AnyQuoteItemBase):
[...]

Default values, Descriptions and Enum

class StripePaymentIntent(BaseModel):
    type: str ="stripe.payment_intent"
    id: str = Field(description="Stripe payment intent id")
    amount_capturable: int | None = Field(description="Amount capturable for this payment intent", default=None)
    currency: CurrencyEnum = Field(description="Currency of the payment intent", default=CurrencyEnum.GBP)
    customer: str | None = None
    client_secret: str

Documenting dictionary with examples

class BillingMethodCreate(BaseModel):
[...]
    metadata: BillingTypeMetadata | None = Field(
        default_factory=dict,
        alias="_metadata",
    )
    model_config = ConfigDict(
        extra="forbid"
    )

Dictionary options are documented in a different class called BillingTypeMetadata
metadata is a reserved word so we use the _metadata alias

The dictionary fields and validation:

class BillingTypeMetadata(BaseModel):
    stripe_customer_id: str | None = Field(default=None, description="Stripe customer id")
    stripe_customer: StripeCustomer | None = Field(default=None, description="Stripe Full Customer Object")
    credit_balance_min: int | None = Field(default=0, description="Minimum credit balance")
    subscription_id: str | None = Field(default=None, description="Stripe subscription id")
    subscription: dict | None = Field(default=None, description="Stripe subscription object")
    subscription_interval: Literal["day", "week", "month", "year"] | None = Field(default=None, description="Subscription interval")
    subscription_interval_count: int | None = Field(default=None, description="Subscription interval count")
    subscription_amount: int | None = Field(default=0, description="The 'monthly' fee for the subscription.")
    subscription_collection_method: Literal["charge_automatically", "send_invoice"] = Field(default="charge_automatically")
    model_config = ConfigDict(
        extra="forbid"
    )

Custom authentication

Use dependencies to add custom code to check authentication and authorization.

@router.get("/{quote_id}", dependencies=[Permission("get", QuoteRepository.factory)])
def get_quote(
        quote_id: UUID,
        quote: model.Quote = Permission("view", QuoteRepository.get),
        db: Session = Depends(database.get_db)
) -> schemas.Quote:

Object ACL or row level permissions with FastAPI and SQLAlchemy

Project: https://github.com/holgi/fastapi-permissions

Route setup

@router.get("/{quote_id}")
def get_quote(
        quote_id: UUID,
        quote: model.Quote = Permission("view", QuoteRepository.get),
        db: Session = Depends(database.get_db)
) -> schemas.Quote:
    return quote

from uuid import UUID
from fastapi_permissions import Allow, Any

SQLAlchemy object

class Quote(Base):
    quote_id: UUID
    owner: UUID
    organization_id: UUID
    
    def __acl__(self):
        acl = []
        if self.owner:
            acl.append((Allow, f"user:{self.owner}", "view"))
            acl.append((Allow, f"user:{self.owner}", "edit"))
            
        if self.organization_id:
            acl.append((Allow, f"organization:{self.organization_id}", "view"))
            acl.append((Allow, f"organization:{self.organization_id}", "accept"))            
        return acl