Fundamentals~ 14 min read

Schemas

Schemas are fundamental to building robust and well-documented APIs with Flama. They provide a formal definition for the structure of data that your application expects to receive and send. By defining schemas, you ensure data integrity, enable automatic validation, and facilitate clear communication of your API's data requirements, both for developers and for automated tooling like API documentation generators.

What are schemas?

In the context of Flama, a Schema is a declaration that describes the expected format of data. This typically applies to request bodies sent to your API (e.g., the JSON payload in a POST or PUT request), and response bodies sent from your API.

Schemas define the fields data should contain, their types (e.g., string, integer, boolean, nested objects, lists), whether they are required or optional, and any validation rules they must adhere to (e.g., minimum/maximum values, string patterns).

Why are they important?

Data validation: Schemas allow Flama to automatically validate incoming request data. If data doesn't conform to the defined schema, an appropriate error response (typically HTTP 422) is returned to the client, preventing invalid data from reaching your business logic.
Serialisation and deserialisation: They define how application data objects are converted to and from wire formats like JSON for request and response bodies.
API documentation: Schemas are a cornerstone of API documentation. Flama uses them to generate detailed OpenAPI specifications, which describe the expected request and response bodies for each endpoint, making your API self-documenting.
Clear contracts: They establish a clear contract between the client and the server regarding data exchange, reducing misunderstandings and integration issues.
Developer experience: Working with well-defined data structures improves code clarity and maintainability.

The main virtue schemas bring is enabling reliable and predictable data exchange by providing a single source of truth for data structures, which Flama leverages for validation, (de)serialisation, and documentation.

Schema libraries in Flama

Flama is designed to be flexible and supports multiple popular Python schema and data validation libraries. Out of the box, it provides adapters for:

Pydantic
Typesystem
Marshmallow

While Flama can work with any of these, only one schema library adapter can be active at a time within a single Flama application instance. The schema library is determined as follows:

Default automatic detection: When the flama.schemas module is first imported (which happens when you import flama.Flama or flama.schemas itself), it attempts to find an installed and supported schema library. It checks for Pydantic, then Typesystem, then Marshmallow. The first one it finds that has a corresponding Flama adapter implemented is set up as the global default. If none of these are installed or have a valid Flama adapter, a flama.exceptions.DependencyNotInstalled error is raised.
Explicit configuration via Flama(schema_library=...): You can explicitly set the schema library for your application (and thus globally reconfigure it for the process if it's different from the auto-detected one) by passing the schema_library parameter when instantiating the Flama class:
# You can also specify either "pydantic" or "marshmallow"app = flama.Flama(schema_library="typesystem")
When the Flama application is initialised, if the schema_library parameter is provided, it instructs the internal SchemaModule to re-initialise Flama's global schema handling system (flama.schemas.adapter, flama.schemas.schemas, etc.) to use the specified library. This makes it the active schema library for all schema operations thereafter within that Python process.

It is generally recommended to explicitly set the schema_library in your Flama application constructor if you intend to use a specific library other than what might be auto-detected, or to ensure consistent behaviour across different environments.

Defining and utilising schemas

Regardless of the chosen library, the general workflow for using schemas in Flama follows a similar pattern.

Defining schema structures

You define your data structures as classes that inherit from a base class provided by the chosen schema library (e.g., pydantic.BaseModel, typesystem.Schema, marshmallow.Schema). Within these classes, you declare fields with their respective types and any validation constraints.

Request body validation and data access

To validate an incoming request body against a schema, you type hint the relevant parameter in your route handler function using typing.Annotated with flama.schemas.Schema and flama.schemas.SchemaMetadata, specifying your schema class. Flama will use the active schema adapter to:

Parse the request body (e.g., from JSON).
Validate the parsed data against your schema class.
If valid, provide the validated data as a Python dictionary to your handler function parameter.
If invalid, automatically return an HTTP 400 Bad Request error response with details about the validation failures.

Crucially, the handler receives the validated data as a dictionary, not (by default) a direct instance of your schema class (e.g., not a Pydantic model instance). If you need an instance of your schema model within the handler (for example, to use its methods, for more complex data manipulation, or simply for more structured attribute access), you would typically instantiate it yourself from this validated dictionary:

async def create_item(data: t.Annotated[  schemas.Schema,  schemas.SchemaMetadata(MyPydanticItemSchema)]):     # item_data is a dictionary here, already validated by MyPydanticItemSchema     # If you need a MyPydanticItemSchema instance:     item_instance = MyPydanticItemSchema(**item_data)     # Now you can use item_instance.some_method() or item_instance.some_field     ...

File uploads

Schemas are not restricted to data that can be expressed in JSON. A schema field may also hold a file, which lets an endpoint receive a multipart/form-data body describing an upload together with whatever metadata accompanies it. In Flama a file within a schema is declared with flama.schemas.fields.File, which always resolves to the file field of whichever schema library is active, and renders in the OpenAPI document as a binary string, so the interactive documentation offers a file picker for it.

Pydantic derives everything from the annotation, so there File is a type to annotate with:

import typing as t
import pydantic
from flama import Flama, schemas, types
class ModelUpload(pydantic.BaseModel):    file: schemas.fields.File    thumbnail: schemas.fields.File | None = None    name: str    version: int

Typesystem and Marshmallow build their schemas from explicit field objects, so there the same name is a field to instantiate:

import marshmallowimport typesystem
from flama import schemas
# TypesystemModelUpload = typesystem.Schema(    title="ModelUpload",    fields={        "file": schemas.fields.File(),        "name": typesystem.fields.String(),        "version": typesystem.fields.Integer(),    },)
# Marshmallowclass ModelUpload(marshmallow.Schema):    file = schemas.fields.File(required=True)    name = marshmallow.fields.String(required=True)    version = marshmallow.fields.Integer(required=True)

The endpoint declares the schema exactly as it would for a JSON body. Remember that the handler receives a validated dictionary, so the file is read from data["file"], and its contents are retrieved with an awaited read:

@app.route("/models/", methods=["POST"])async def upload_model(    data: t.Annotated[types.Schema, types.SchemaMetadata(ModelUpload)],) -> t.Annotated[types.Schema, types.SchemaMetadata(UploadResult)]:    """    tags:        - Models    summary:        Upload a model    description:        Stores a serialised model along with the metadata describing it.    responses:        200:            description:                Model stored.    """    contents = await data["file"].read()    return {"filename": data["file"].filename, "size": len(contents), "name": data["name"]}

Whichever library declares the schema, the value handed to the endpoint is a flama.http.UploadFile. Alongside read, it exposes the filename and content_type declared in the part headers, a seek method to rewind the stream, and the headers of the part itself.

When an endpoint receives nothing but a file, a schema adds little, so a bare UploadFile parameter is accepted as the whole body. Flama documents it as a multipart body with a single binary property named after the parameter:

from flama.http import UploadFile
@app.route("/avatar/", methods=["POST"])async def upload_avatar(avatar: UploadFile) -> t.Annotated[types.Schema, types.SchemaMetadata(UploadResult)]:    """    tags:        - Users    summary:        Upload an avatar    description:        Replaces the avatar picture of the current user.    responses:        200:            description:                Avatar updated.    """    return {"filename": avatar.filename, "size": len((await avatar.read()))}

Nothing about the request is specific to Flama: clients send an ordinary multipart/form-data body, where each schema field corresponds to a part of the same name.

$ curl -X POST http://localhost:8000/models/ \    -F "[email protected]" \    -F "name=my-model" \    -F "version=3"

{ "filename": "model.pkl", "size": 20480, "name": "my-model" }

Working with multipart bodies

Multipart is a flatter format than JSON, and there are three consequences worth knowing before designing a schema around it.

The first is that every non-file part travels as text. This is not something you need to compensate for, because each schema library coerces those strings into the declared types, so a version field declared as an integer arrives as an int and a boolean field sent as true arrives as a bool.

The second is that a repeated part name is how a collection is expressed. Sending several parts that share a name produces a list, which is what a field such as list[schemas.fields.File] expects:

$ curl -X POST http://localhost:8000/gallery/ \    -F "[email protected]" \    -F "[email protected]"

WARNING

A list field needs at least two parts to be recognised as a list. When a single part is sent, the form carries no information distinguishing it from an ordinary scalar, and validation of a list-typed field fails. Where a one-element upload has to be accepted, declare a field that tolerates both shapes, such as schemas.fields.File | list[schemas.fields.File] with Pydantic, and normalise it inside the handler.

The third is that nesting cannot be expressed at all. A part name maps to a top-level field, so a schema containing another schema has no valid multipart encoding, and sending the nested object as a JSON string fails validation. Where an upload has to be accompanied by structured metadata, either flatten the schema into individual fields, or declare the field as a string and parse it inside a validator.

Large uploads

Holding an entire upload in memory does not scale, so Flama streams every part and keeps it in memory only while it stays below a threshold, one mebibyte by default. Past that point the part is written to a temporary file, and UploadFile.path reports where it lives. Reading is identical either way, so an endpoint never needs to know which of the two happened. The temporary file is removed once the response has been sent.

Uploads are unbounded unless a limit is configured. To bound them, register a RequestDataComponent with the limits appropriate to the application, and any request exceeding them is rejected with an HTTP 413 Content Too Large response before the body is fully read:

from flama import Flamafrom flama.schemas.components import RequestDataComponent
app = Flama(    openapi={"info": {"title": "Model Registry", "version": "1.0.0"}},    components=[        RequestDataComponent(            max_file_size=50 * 1024 * 1024,  # Largest single upload accepted            max_body_size=200 * 1024 * 1024,  # Largest request body accepted            spool_threshold=1024 * 1024,  # Point at which a part moves to disk            max_files=10,  # Largest number of file parts accepted        )    ],)

Response model definition

To define the structure of a response and ensure your handler's return value is serialised correctly (and accurately documented in the OpenAPI schema), you use a schema as the return type annotation of your route handler function. This also uses typing.Annotated with flama.schemas.Schema and flama.schemas.SchemaMetadata.

In particular, Flama will:

Take the object returned by your handler (which should be a dictionary conforming to the schema, or an instance of your schema model if the adapter can serialise it directly).
Serialise it to the appropriate format (usually JSON) according to the specified schema.
Include the schema definition in the OpenAPI documentation for that endpoint's response.

The response schema is solely determined by the return type annotation.

async def item_detals(id: 'str') -> t.Annotated[  schemas.Schema,  schemas.SchemaMetadata(MyPydanticItemSchema)]:     ...

The role of `Schema` and `SchemaMetadata`

As we've seen so far, we need the use of the following tools from Flama to be able to use schemas for validation and (de)serialisation:

schemas.Schema: This is an alias for dict[str, t.Any]. When used in a type hint like t.Annotated[schemas.Schema, schemas.SchemaMetadata(YourSchemaClass)] for a request body, it means that the handler will receive a dictionary, which Flama has validated against YourSchemaClass.
schemas.SchemaMetadata(schema: t.Any, partial: bool = False): This dataclass, used as the second argument within typing.Annotated, provides metadata to Flama.
- schema: The actual schema class you defined (e.g., your Puppy Pydantic model class). This tells Flama which schema definition to use for validation or serialisation, and for documentation.
- partial: A boolean (default False). If True, it indicates that a partial update is allowed (some fields might be missing). Support and behaviour depend on the schema library adapter.

Schema registration

To ensure your schemas are correctly represented in the global OpenAPI documentation (specifically in the components.schemas section, allowing them to be referenced by a consistent name), you should register them with your Flama application instance. To this purpose, we use app.schema.register_schema(name: str, schema: type) with name being the string name under which this schema will be registered in the OpenAPI document (e.g., "Puppy"), and schema the schema class itself.

class X(pydantic.BaseModel):    input: list[t.Any] = pydantic.Field(title="input", description="Model input")
class Y(pydantic.BaseModel):    output: list[t.Any] = pydantic.Field(title="output", description="Model output")
app.schema.register_schema("X", X)app.schema.register_schema("Y", Y)

Examples

The following examples demonstrate defining and using a Puppy schema for a "Puppy Register" API with Pydantic, Typesystem, and Marshmallow. Each handler receives and returns dictionaries.

Pydantic

Pydantic is the default schema library if installed.

import typing as timport pydanticimport uuid
import flamafrom flama import Flama, schemas
PUPPIES_DB_PYDANTIC: list[dict[str, t.Any]] = []
class Puppy(pydantic.BaseModel):    id: uuid.UUID    name: str    age: int
    @pydantic.field_validator("age")    @classmethod    def age_validation(cls, v: int) -> int:        if v < 0:            raise ValueError("Age must be a non-negative number.")        if v > 30:            raise ValueError("Age seems too high for a puppy.")        return v
class PuppyCreatePayload(pydantic.BaseModel):    name: str    age: int
    @pydantic.field_validator("age")    @classmethod    def age_validation(        cls, v: int    ) -> int:  # Re-apply validation if needed for input schema        if v < 0:            raise ValueError("Age must be a non-negative number.")        if v > 30:            raise ValueError("Age seems too high for a puppy.")        return v
app = Flama(    openapi={"info": {"title": "Puppy Register (Pydantic)", "version": "0.1.0"}},    schema_library="pydantic",)
app.schema.register_schema(name="Puppy", schema=Puppy)app.schema.register_schema(name="PuppyCreationPayload", schema=PuppyCreatePayload)
PuppyListResponse = t.Annotated[    list[schemas.Schema],    schemas.SchemaMetadata(Puppy),]PuppyDetailResponse = t.Annotated[    schemas.Schema,    schemas.SchemaMetadata(Puppy),]PuppyCreateRequest = t.Annotated[    schemas.Schema,    schemas.SchemaMetadata(PuppyCreatePayload),]
@app.get("/puppies/")async def list_puppies_pydantic(name: t.Optional[str] = None) -> PuppyListResponse:    result = [p for p in PUPPIES_DB_PYDANTIC if name is None or p.get("name") == name]    return result
@app.post("/puppies/")async def create_puppy_pydantic(    puppy_input_dict: PuppyCreateRequest,) -> PuppyDetailResponse:    # puppy_input_dict is a validated dict matching PuppyCreatePayload    puppy = {        "name": puppy_input_dict["name"],        "age": puppy_input_dict["age"],        "id": uuid.uuid4(),    }
    # Explicit validation of the composed object before returning (optional):    Puppy.model_validate(puppy)
    PUPPIES_DB_PYDANTIC.append(puppy)
    return puppy
if __name__ == "__main__":    flama.run(flama_app=app, server_host="0.0.0.0", server_port=8000)

Typesystem

Ensure Typesystem is installed and selected via schema_library="typesystem".

import typing as timport uuid
import flamafrom flama import Flama, schemas
import typesystem
PUPPIES_DB_TYPESYSTEM: list[dict[str, t.Any]] = []
puppy_output_schema = typesystem.Schema(    fields={        "id": typesystem.String(title="ID"),        "name": typesystem.String(title="Name", max_length=100),        "age": typesystem.Integer(title="Age", minimum=0, maximum=30),    })
puppy_create_schema = typesystem.Schema(    fields={        "name": typesystem.String(title="Name", max_length=100, allow_null=False),        "age": typesystem.Integer(title="Age", minimum=0, maximum=30, allow_null=False),    })
app = Flama(    openapi={"info": {"title": "Puppy Register (Typesystem)", "version": "0.1.0"}},    schema_library="typesystem",)
app.schema.register_schema(name="Puppy", schema=puppy_output_schema)app.schema.register_schema(name="PuppyCreationPayload", schema=puppy_create_schema)
PuppyListResponse = t.Annotated[    list[schemas.Schema],    schemas.SchemaMetadata(puppy_output_schema),]PuppyDetailResponse = t.Annotated[    schemas.Schema,    schemas.SchemaMetadata(puppy_output_schema),]PuppyCreateRequest = t.Annotated[    schemas.Schema,    schemas.SchemaMetadata(puppy_create_schema),]
@app.get("/puppies/")async def list_puppies_typesystem(name: t.Optional[str] = None) -> PuppyListResponse:    result = [p for p in PUPPIES_DB_TYPESYSTEM if name is None or p.get("name") == name]    return result
@app.post("/puppies/")async def create_puppy_typesystem(    puppy_input_dict: PuppyCreateRequest,) -> PuppyDetailResponse:    puppy = {        "name": puppy_input_dict["name"],        "age": puppy_input_dict["age"],        "id": str(uuid.uuid4()),    }
    # Explicit validation of the composed object before returning (optional):    puppy_output_schema.validate(puppy)
    PUPPIES_DB_TYPESYSTEM.append(puppy)
    return puppy

if __name__ == "__main__":    flama.run(flama_app=app, server_host="0.0.0.0", server_port=8000)

Marshmallow

Ensure Marshmallow is installed and selected via schema_library="marshmallow".

import typing as timport uuid
import marshmallowfrom marshmallow import fields, validate
import flamafrom flama import Flama, schemas
PUPPIES_DB_MARSHMALLOW: list[dict[str, t.Any]] = []
class PuppyOutput(marshmallow.Schema):    id = fields.UUID(dump_only=True)  # UUID field, for output only    name = fields.Str(required=True, validate=validate.Length(min=1, max=100))    age = fields.Int(        required=True,        validate=validate.Range(min=0, max=30, error="Age must be between 0 and 30."),    )
    class Meta:        ordered = True
class PuppyCreatePayload(marshmallow.Schema):    name = fields.Str(required=True, validate=validate.Length(min=1, max=100))    age = fields.Int(        required=True,        validate=validate.Range(min=0, max=30, error="Age must be between 0 and 30."),    )
    class Meta:        ordered = True
app = Flama(    openapi={"info": {"title": "Puppy Register (Marshmallow)", "version": "0.1.0"}},    schema_library="marshmallow",)
app.schema.register_schema(name="PuppyMM", schema=PuppyOutput)app.schema.register_schema(name="PuppyCreationPayloadMM", schema=PuppyCreatePayload)
PuppyListResponse = t.Annotated[    list[schemas.Schema],    schemas.SchemaMetadata(PuppyOutput),]PuppyDetailResponse = t.Annotated[    schemas.Schema,    schemas.SchemaMetadata(PuppyOutput),]PuppyCreateRequest = t.Annotated[    schemas.Schema,    schemas.SchemaMetadata(PuppyCreatePayload),]
@app.get("/puppies/")async def list_puppies_marshmallow(name: t.Optional[str] = None) -> PuppyListResponse:    result = [        p for p in PUPPIES_DB_MARSHMALLOW if name is None or p.get("name") == name    ]    return result
@app.post("/puppies/")async def create_puppy_marshmallow(    puppy_input: PuppyCreateRequest,) -> PuppyDetailResponse:    puppy = {        "name": puppy_input["name"],        "age": puppy_input["age"],        "id": uuid.uuid4(),    }
    # Explicit validation of the composed object before returning (optional):    PuppyOutput().validate(puppy)
    PUPPIES_DB_MARSHMALLOW.append(puppy)
    return puppy
if __name__ == "__main__":    flama.run(flama_app=app, server_host="0.0.0.0", server_port=8000)

Introduction

Getting Started

Fundamentals

Flama CLI

Advanced Topics

Predictive AI

Generative AI

Domain driven design

Contributing

Schemas

What are schemas?

Schema libraries in Flama

Defining and utilising schemas

Defining schema structures

Request body validation and data access

File uploads

Working with multipart bodies

Large uploads

Response model definition

The role of `Schema` and `SchemaMetadata`

Schema registration

Examples

Pydantic

Typesystem

Marshmallow

Introduction

Getting Started

Fundamentals

Flama CLI

Advanced Topics

Predictive AI

Generative AI

Domain driven design

Contributing

Schemas

What are schemas?

Schema libraries in Flama

Defining and utilising schemas

Defining schema structures

Request body validation and data access

File uploads

Working with multipart bodies

Large uploads

Response model definition

The role of Schema and SchemaMetadata

Schema registration

Examples

Pydantic

Typesystem

Marshmallow

The role of `Schema` and `SchemaMetadata`