in python

Pydantic / FastAPI quick notes

Personal quick notes to work with FastAPI / Pydantic.

  1. Pydantic acceptinc multiple payloads with discrimination field
  2. Default values, documentation, description and examples using Field
  3. 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

Write a Comment

Comment