Project 6: Generic Model Library

Project 6: Generic Model Library

Build a library of generic Pydantic models–paginated responses, API envelopes, result types–that work with any data type while maintaining full type safety.

Learning Objectives

By completing this project, you will:

  1. Master Python generics with TypeVar - Understand how type variables enable reusable, type-safe code
  2. Create generic Pydantic BaseModel subclasses - Build models that work with any data type while maintaining validation
  3. Understand bounded type variables - Constrain type variables with bound for safer generic types
  4. Implement covariance and contravariance - Know when to use covariant=True or contravariant=True
  5. Use runtime type introspection - Apply get_origin and get_args to inspect generic types at runtime
  6. Integrate generic models with FastAPI - Create reusable response wrappers that generate correct OpenAPI schemas

Deep Theoretical Foundation

The Problem Generics Solve

Consider building an API that returns paginated responses for different resources:

# Without generics: Massive duplication
class PaginatedUserResponse(BaseModel):
    items: list[User]
    total: int
    page: int
    per_page: int

class PaginatedOrderResponse(BaseModel):
    items: list[Order]
    total: int
    page: int
    per_page: int

class PaginatedProductResponse(BaseModel):
    items: list[Product]
    total: int
    page: int
    per_page: int

# ... repeated for every entity

This duplication violates DRY (Don’t Repeat Yourself) and makes maintenance painful. Generics solve this:

# With generics: Single reusable definition
T = TypeVar('T')

class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    per_page: int

# Usage - fully type-safe!
UserPage = PaginatedResponse[User]
OrderPage = PaginatedResponse[Order]
ProductPage = PaginatedResponse[Product]

Python Generics Fundamentals

TypeVar: The Type Variable

A TypeVar is a placeholder for a type that will be specified later. Think of it as a variable that holds a type instead of a value:

Regular Variable:     x = 5           # x holds the value 5
Type Variable:        T = TypeVar('T') # T holds a type (like int, str, User)

Creating type variables:

from typing import TypeVar

# Basic type variable - can be any type
T = TypeVar('T')

# Type variable with constraints - can only be specific types
Number = TypeVar('Number', int, float)  # Only int or float

# Type variable with bound - must be subclass of bound
from pydantic import BaseModel
ModelT = TypeVar('ModelT', bound=BaseModel)  # Must be BaseModel or subclass

How TypeVar Works at Runtime vs Static Analysis

Static Analysis (mypy, Pyright)    Runtime (Python Interpreter)
================================   ============================

T = TypeVar('T')                   T = TypeVar('T')
                                   # T is a TypeVar object

class Box(Generic[T]):             class Box(Generic[T]):
    value: T                           value: T
                                       # T is still a TypeVar

box: Box[int] = Box(value=5)       box = Box(value=5)
                                   # Type info mostly erased!

reveal_type(box.value)  # int      type(box.value)  # int (from value)

Key insight: Python’s type hints are largely erased at runtime. Pydantic is special because it preserves and uses type information at runtime for validation.

Generic Class Inheritance Pattern

                    Generic[T]
                        |
                        |  (inherit with Generic[T])
                        v
              +---------+----------+
              |                    |
              v                    v
        BaseModel              Other Classes
              |
              |  (combine inheritance)
              v
    MyGenericModel(BaseModel, Generic[T])
              |
              |  (specialize with concrete type)
              v
    MyGenericModel[User]  # Concrete type

Generic BaseModel Inheritance

When creating a generic Pydantic model, you must inherit from both BaseModel and Generic[T]:

from typing import TypeVar, Generic
from pydantic import BaseModel

T = TypeVar('T')

class Wrapper(BaseModel, Generic[T]):
    data: T
    metadata: dict = {}

How Pydantic handles this:

1. Class Definition
   Wrapper(BaseModel, Generic[T])
       |
       v
2. Pydantic Metaclass Processing
   - Detects Generic[T] in bases
   - Records type parameters
   - Prepares for parameterization
       |
       v
3. Parameterization: Wrapper[User]
   - Creates specialized class
   - Substitutes T with User
   - Generates schema for User
       |
       v
4. Validation
   - Uses User schema for 'data' field
   - Full Pydantic validation applies

Bounded Type Variables

A bound constrains what types can be used with a TypeVar:

from pydantic import BaseModel
from typing import TypeVar, Generic

# Without bound - T can be ANY type
T = TypeVar('T')

class Container(BaseModel, Generic[T]):
    item: T

    def describe(self) -> str:
        # Problem: T could be anything, so we can't call methods on it
        # return self.item.some_method()  # Type error!
        return str(self.item)

# With bound - T must be BaseModel or subclass
ModelT = TypeVar('ModelT', bound=BaseModel)

class ModelContainer(BaseModel, Generic[ModelT]):
    item: ModelT

    def get_fields(self) -> list[str]:
        # Safe: We know item is a BaseModel, so model_fields exists
        return list(self.item.model_fields.keys())

Visual comparison:

T = TypeVar('T')                    ModelT = TypeVar('ModelT', bound=BaseModel)

Can be:                             Can be:
  - int                               - BaseModel (itself)
  - str                               - User (if User inherits BaseModel)
  - list                              - Order (if Order inherits BaseModel)
  - User
  - Any type!                       Cannot be:
                                      - int
                                      - str
                                      - dict
                                      - Classes not inheriting BaseModel

Covariance and Contravariance

These concepts describe how generic types relate to each other when their type parameters are related by inheritance.

Covariance (Output Position)

If Dog is a subtype of Animal, then Container[Dog] is a subtype of Container[Animal]:

T_co = TypeVar('T_co', covariant=True)

class Reader(Generic[T_co]):
    """A type that only PRODUCES values of type T"""
    def read(self) -> T_co:
        ...

# Dog is subtype of Animal
# Therefore Reader[Dog] is subtype of Reader[Animal]

def process_reader(r: Reader[Animal]) -> None:
    animal = r.read()  # We get an Animal (or subtype)

dog_reader: Reader[Dog] = ...
process_reader(dog_reader)  # OK! Reader[Dog] is subtype of Reader[Animal]

Inheritance:     Dog --> Animal
                  |        |
Covariance:      v        v
             Reader[Dog] --> Reader[Animal]
             (subtype)       (supertype)

Contravariance (Input Position)

The subtype relationship is reversed for types that CONSUME values:

T_contra = TypeVar('T_contra', contravariant=True)

class Writer(Generic[T_contra]):
    """A type that only CONSUMES values of type T"""
    def write(self, value: T_contra) -> None:
        ...

# Dog is subtype of Animal
# Therefore Writer[Animal] is subtype of Writer[Dog]

def use_writer(w: Writer[Dog]) -> None:
    w.write(Dog())

animal_writer: Writer[Animal] = ...
use_writer(animal_writer)  # OK! Writer[Animal] can accept Dog

Inheritance:     Dog --> Animal
                  |        |
Contravariance:  v        v
             Writer[Animal] --> Writer[Dog]
             (subtype)          (supertype)
             (reversed!)

Invariance (Default)

By default, TypeVar is invariant–no subtype relationship:

T = TypeVar('T')  # Invariant

class Container(Generic[T]):
    def get(self) -> T: ...
    def set(self, value: T) -> None: ...

# Container[Dog] is NOT a subtype of Container[Animal]
# Container[Animal] is NOT a subtype of Container[Dog]

Practical Pydantic Example

from typing import TypeVar, Generic, Sequence
from pydantic import BaseModel

# For response models (output), use covariance
T_co = TypeVar('T_co', covariant=True, bound=BaseModel)

class ResponseWrapper(BaseModel, Generic[T_co]):
    """Returns data - covariant makes sense for API responses"""
    success: bool
    data: T_co

# Note: Pydantic models are typically invariant by default
# because they both read and write data. Covariance is
# mainly useful for abstract/read-only interfaces.

Runtime Type Introspection

At runtime, you often need to inspect generic types. Python’s typing module provides get_origin and get_args:

from typing import get_origin, get_args, List, Dict, Optional, Union

# get_origin: Returns the base type
get_origin(List[int])           # list
get_origin(Dict[str, int])      # dict
get_origin(Optional[str])       # typing.Union
get_origin(int)                 # None (not generic)

# get_args: Returns the type arguments
get_args(List[int])             # (int,)
get_args(Dict[str, int])        # (str, int)
get_args(Optional[str])         # (str, NoneType)
get_args(int)                   # () (empty tuple)

Diagram of what these functions return:

Type Expression          get_origin()      get_args()
---------------          ------------      ----------
List[int]                list              (int,)
Dict[str, User]          dict              (str, User)
Optional[Order]          Union             (Order, NoneType)
Union[int, str, None]    Union             (int, str, NoneType)
Tuple[int, str, float]   tuple             (int, str, float)
Set[Product]             set               (Product,)
Callable[[int], str]     collections.abc.  (int, str)
                         Callable

Using Introspection with Pydantic Generics

from typing import TypeVar, Generic, get_origin, get_args, Type
from pydantic import BaseModel

T = TypeVar('T')

class Wrapper(BaseModel, Generic[T]):
    data: T

    @classmethod
    def get_inner_type(cls) -> Type:
        """Get the actual type T was parameterized with."""
        # For Wrapper[User], this returns User
        for base in cls.__orig_bases__:
            origin = get_origin(base)
            if origin is Wrapper or (origin and issubclass(origin, Wrapper)):
                args = get_args(base)
                if args:
                    return args[0]
        return None

# Usage
UserWrapper = Wrapper[User]
print(UserWrapper.get_inner_type())  # <class 'User'>

Generic Types with FastAPI

FastAPI fully supports Pydantic generic models in response types:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import TypeVar, Generic

T = TypeVar('T')

class APIResponse(BaseModel, Generic[T]):
    success: bool
    data: T | None = None
    message: str | None = None

class User(BaseModel):
    id: int
    name: str

app = FastAPI()

@app.get("/users/{user_id}", response_model=APIResponse[User])
async def get_user(user_id: int):
    user = User(id=user_id, name="John")
    return APIResponse(success=True, data=user)

The OpenAPI schema correctly reflects the nested structure:

{
  "APIResponse_User_": {
    "type": "object",
    "properties": {
      "success": {"type": "boolean"},
      "data": {
        "anyOf": [
          {"$ref": "#/components/schemas/User"},
          {"type": "null"}
        ]
      },
      "message": {
        "anyOf": [
          {"type": "string"},
          {"type": "null"}
        ]
      }
    },
    "required": ["success"]
  }
}

Nested Generics

You can compose generic types for complex structures:

T = TypeVar('T')
E = TypeVar('E')

class Result(BaseModel, Generic[T, E]):
    """Either success with data T, or failure with error E"""
    success: bool
    data: T | None = None
    error: E | None = None

class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int

# Nested generic: API returns paginated results that can fail
@app.get("/users", response_model=Result[PaginatedResponse[User], str])
async def list_users(page: int = 1):
    try:
        users = get_users(page)
        paginated = PaginatedResponse(items=users, total=100, page=page)
        return Result(success=True, data=paginated)
    except Exception as e:
        return Result(success=False, error=str(e))

Type structure visualization:

Result[PaginatedResponse[User], str]
    |
    +-- success: bool
    |
    +-- data: PaginatedResponse[User] | None
    |       |
    |       +-- items: list[User]
    |       |       |
    |       |       +-- User instances
    |       |
    |       +-- total: int
    |       +-- page: int
    |
    +-- error: str | None

Project Specification

Functional Requirements

Build a library of generic Pydantic models that can be used across any Python API project:

  1. APIResponse[T] - Standard API response envelope
    • Wrap any data type with success/error status
    • Include optional message and metadata
    • Provide factory methods for success/error cases
  2. PaginatedResponse[T] - Paginated list wrapper
    • Support any item type
    • Include pagination metadata (page, per_page, total, pages)
    • Provide helper properties (has_next, has_prev)
    • Include factory method to paginate sequences
  3. Result[T, E] - Rust-style Result type
    • Either success with data of type T
    • Or failure with error of type E
    • Monadic operations (map, flat_map)
  4. Optional[T] Enhancement - Rich optional wrapper
    • Better than None for explicit absence
    • Support default value on unwrap
    • Chain operations safely
  5. BatchResponse[T] - Bulk operation results
    • Track individual item successes/failures
    • Aggregate statistics
    • Maintain order correspondence

Model Contracts

# APIResponse[T]
class APIResponse(BaseModel, Generic[T]):
    success: bool = True
    data: T | None = None
    message: str | None = None
    errors: list[str] | None = None
    meta: dict | None = None
    timestamp: datetime

    @classmethod
    def ok(cls, data: T, message: str = None) -> "APIResponse[T]": ...

    @classmethod
    def error(cls, message: str, errors: list[str] = None) -> "APIResponse[T]": ...

# PaginatedResponse[T]
class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int = Field(ge=1)
    per_page: int = Field(ge=1, le=100)

    @property
    def pages(self) -> int: ...

    @property
    def has_next(self) -> bool: ...

    @property
    def has_prev(self) -> bool: ...

    @classmethod
    def from_sequence(cls, items: Sequence[T], page: int, per_page: int) -> "PaginatedResponse[T]": ...

# Result[T, E]
class Result(BaseModel, Generic[T, E]):
    _is_ok: bool
    _value: T | None
    _error: E | None

    @classmethod
    def ok(cls, value: T) -> "Result[T, E]": ...

    @classmethod
    def err(cls, error: E) -> "Result[T, E]": ...

    def is_ok(self) -> bool: ...
    def is_err(self) -> bool: ...
    def unwrap(self) -> T: ...  # Raises if error
    def unwrap_or(self, default: T) -> T: ...
    def map(self, fn: Callable[[T], U]) -> "Result[U, E]": ...

Solution Architecture

Project Structure

generic-models/
    src/
        generic_models/
            __init__.py
            api_response.py       # APIResponse[T]
            paginated.py          # PaginatedResponse[T]
            result.py             # Result[T, E]
            batch.py              # BatchResponse[T]
            optional.py           # OptionalValue[T]
            types.py              # Common TypeVars
            utils.py              # Runtime introspection helpers
    tests/
        __init__.py
        test_api_response.py
        test_paginated.py
        test_result.py
        test_fastapi_integration.py
    examples/
        fastapi_app.py            # Example FastAPI integration
        basic_usage.py            # Simple usage examples
    pyproject.toml
    README.md

Component Diagram

+-----------------------------------------------------------------+
|                     generic_models Package                       |
+-----------------------------------------------------------------+
|                                                                  |
|  +------------------+    +-------------------+                   |
|  |   types.py       |    |     utils.py      |                   |
|  +------------------+    +-------------------+                   |
|  | T = TypeVar('T') |    | get_inner_type()  |                   |
|  | E = TypeVar('E') |    | resolve_generic() |                   |
|  | ModelT = TypeVar |    | is_generic()      |                   |
|  |   (bound=...)    |    |                   |                   |
|  +--------+---------+    +---------+---------+                   |
|           |                        |                             |
|           v                        v                             |
|  +------------------+    +-------------------+                   |
|  | api_response.py  |    |   paginated.py    |                   |
|  +------------------+    +-------------------+                   |
|  | APIResponse[T]   |    | PaginatedResponse |                   |
|  | - ok()           |    | [T]               |                   |
|  | - error()        |    | - from_sequence() |                   |
|  +--------+---------+    | - pages property  |                   |
|           |              +-------------------+                   |
|           |                        |                             |
|           v                        v                             |
|  +------------------+    +-------------------+                   |
|  |   result.py      |    |    batch.py       |                   |
|  +------------------+    +-------------------+                   |
|  | Result[T, E]     |    | BatchResponse[T]  |                   |
|  | - ok() / err()   |    | - BatchItem[T]    |                   |
|  | - map()          |    | - statistics      |                   |
|  | - unwrap()       |    |                   |                   |
|  +------------------+    +-------------------+                   |
|                                                                  |
+-----------------------------------------------------------------+
                              |
                              v
              +-------------------------------+
              |     FastAPI Integration       |
              +-------------------------------+
              | @app.get(response_model=      |
              |   APIResponse[User])          |
              | - OpenAPI schema generation   |
              | - Automatic serialization     |
              +-------------------------------+

Type Hierarchy

BaseModel (Pydantic)
    |
    +-- Generic[T] (typing)
            |
            +-- APIResponse[T]
            |       |
            |       +-- APIResponse[User]
            |       +-- APIResponse[Order]
            |       +-- APIResponse[PaginatedResponse[Product]]
            |
            +-- PaginatedResponse[T]
            |       |
            |       +-- PaginatedResponse[User]
            |       +-- PaginatedResponse[Order]
            |
            +-- Result[T, E]
            |       |
            |       +-- Result[User, ValidationError]
            |       +-- Result[Order, str]
            |
            +-- BatchResponse[T]
                    |
                    +-- BatchResponse[User]

Phased Implementation Guide

Phase 1: Foundation and TypeVars (1-2 hours)

Goal: Set up the project structure and define reusable type variables.

  1. Create project structure: 
    mkdir -p generic-models/src/generic_models generic-models/tests generic-models/examples
touch generic-models/src/generic_models/__init__.py
touch generic-models/pyproject.toml
  2. Create types.py with common TypeVars: 
    from typing import TypeVar
from pydantic import BaseModel

# Generic type variable - can be any type
T = TypeVar('T')

# Second type variable for Result[T, E]
E = TypeVar('E')

# Third type variable for complex compositions
U = TypeVar('U')

# Bounded type variable - must be BaseModel subclass
ModelT = TypeVar('ModelT', bound=BaseModel)
  3. Create utils.py with introspection helpers: 
    from typing import get_origin, get_args, Type, Any

def get_inner_type(cls: Type) -> Type | None:
    """Extract the type parameter from a parameterized generic."""
    for base in getattr(cls, '__orig_bases__', []):
        args = get_args(base)
        if args:
            return args[0]
    return None

def is_generic_model(cls: Type) -> bool:
    """Check if a class is a parameterized generic Pydantic model."""
    return hasattr(cls, '__pydantic_generic_metadata__')

Checkpoint: Import types.py without errors.

Phase 2: APIResponse[T] (2 hours)

Goal: Create a complete API response wrapper.

  1. Create api_response.py: 
    from datetime import datetime
from typing import Generic
from pydantic import BaseModel, Field
from .types import T

class APIResponse(BaseModel, Generic[T]):
    """Standard API response envelope."""
    success: bool = True
    data: T | None = None
    message: str | None = None
    errors: list[str] | None = None
    meta: dict | None = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)

    @classmethod
    def ok(cls, data: T, message: str | None = None, meta: dict | None = None) -> "APIResponse[T]":
        """Create a success response."""
        return cls(success=True, data=data, message=message, meta=meta)

    @classmethod
    def error(cls, message: str, errors: list[str] | None = None) -> "APIResponse[T]":
        """Create an error response."""
        return cls(success=False, message=message, errors=errors)
  2. Test basic functionality: 
    class User(BaseModel):
    id: int
    name: str

response = APIResponse[User].ok(User(id=1, name="John"))
print(response.model_dump_json(indent=2))

Checkpoint: APIResponse works with concrete types and generates valid JSON.

Phase 3: PaginatedResponse[T] (2 hours)

Goal: Create a pagination wrapper with helpers.

  1. Create paginated.py: 
    from typing import Generic, Sequence
from pydantic import BaseModel, Field, computed_field
from .types import T

class PaginatedResponse(BaseModel, Generic[T]):
    """Generic paginated response."""
    items: list[T]
    total: int
    page: int = Field(ge=1, default=1)
    per_page: int = Field(ge=1, le=100, default=10)

    @computed_field
    @property
    def pages(self) -> int:
        """Total number of pages."""
        if self.per_page <= 0:
            return 0
        return (self.total + self.per_page - 1) // self.per_page

    @computed_field
    @property
    def has_next(self) -> bool:
        """Whether there's a next page."""
        return self.page < self.pages

    @computed_field
    @property
    def has_prev(self) -> bool:
        """Whether there's a previous page."""
        return self.page > 1

    @classmethod
    def from_sequence(
        cls,
        items: Sequence[T],
        page: int = 1,
        per_page: int = 10
    ) -> "PaginatedResponse[T]":
        """Create paginated response from a full sequence."""
        total = len(items)
        start = (page - 1) * per_page
        end = start + per_page
        return cls(
            items=list(items[start:end]),
            total=total,
            page=page,
            per_page=per_page
        )

Checkpoint: Pagination correctly calculates pages and slices items.

Phase 4: Result[T, E] (2-3 hours)

Goal: Implement a Rust-style Result type with monadic operations.

  1. Create result.py: 
    from typing import Generic, Callable, TypeVar
from pydantic import BaseModel, model_validator
from .types import T, E

U = TypeVar('U')

class Result(BaseModel, Generic[T, E]):
    """Rust-style Result type: either Ok(T) or Err(E)."""
    _is_success: bool = True
    value: T | None = None
    error: E | None = None

    @model_validator(mode='after')
    def validate_result(self):
        """Ensure exactly one of value or error is set."""
        has_value = self.value is not None
        has_error = self.error is not None

        if self._is_success and has_error:
            raise ValueError("Success result cannot have error")
        if not self._is_success and has_value:
            raise ValueError("Error result cannot have value")
        return self

    @classmethod
    def ok(cls, value: T) -> "Result[T, E]":
        """Create a success result."""
        return cls(_is_success=True, value=value, error=None)

    @classmethod
    def err(cls, error: E) -> "Result[T, E]":
        """Create an error result."""
        return cls(_is_success=False, value=None, error=error)

    def is_ok(self) -> bool:
        """Check if result is success."""
        return self._is_success

    def is_err(self) -> bool:
        """Check if result is error."""
        return not self._is_success

    def unwrap(self) -> T:
        """Get value or raise if error."""
        if self.is_err():
            raise ValueError(f"Called unwrap on error: {self.error}")
        return self.value

    def unwrap_or(self, default: T) -> T:
        """Get value or return default if error."""
        return self.value if self.is_ok() else default

    def unwrap_err(self) -> E:
        """Get error or raise if success."""
        if self.is_ok():
            raise ValueError(f"Called unwrap_err on success: {self.value}")
        return self.error

    def map(self, fn: Callable[[T], U]) -> "Result[U, E]":
        """Transform the success value."""
        if self.is_ok():
            return Result[U, E].ok(fn(self.value))
        return Result[U, E].err(self.error)

    def map_err(self, fn: Callable[[E], U]) -> "Result[T, U]":
        """Transform the error value."""
        if self.is_err():
            return Result[T, U].err(fn(self.error))
        return Result[T, U].ok(self.value)

Checkpoint: Result type works with map operations.

Phase 5: BatchResponse[T] (2 hours)

Goal: Track results of bulk operations.

  1. Create batch.py: 
    from typing import Generic
from pydantic import BaseModel, computed_field
from .types import T
from .result import Result

class BatchItem(BaseModel, Generic[T]):
    """Single item in a batch operation."""
    index: int
    success: bool
    data: T | None = None
    error: str | None = None

class BatchResponse(BaseModel, Generic[T]):
    """Response for batch/bulk operations."""
    results: list[BatchItem[T]]

    @computed_field
    @property
    def total(self) -> int:
        """Total items processed."""
        return len(self.results)

    @computed_field
    @property
    def succeeded(self) -> int:
        """Count of successful items."""
        return sum(1 for r in self.results if r.success)

    @computed_field
    @property
    def failed(self) -> int:
        """Count of failed items."""
        return sum(1 for r in self.results if not r.success)

    @computed_field
    @property
    def success_rate(self) -> float:
        """Percentage of successful operations."""
        if self.total == 0:
            return 0.0
        return self.succeeded / self.total

    def get_successful(self) -> list[T]:
        """Get all successfully processed items."""
        return [r.data for r in self.results if r.success and r.data is not None]

    def get_failures(self) -> list[tuple[int, str]]:
        """Get all failures as (index, error) tuples."""
        return [(r.index, r.error or "Unknown error")
                for r in self.results if not r.success]

Checkpoint: BatchResponse correctly aggregates statistics.

Phase 6: FastAPI Integration and Testing (3 hours)

Goal: Integrate with FastAPI and create comprehensive tests.

  1. Create examples/fastapi_app.py: 
    from fastapi import FastAPI, Query, HTTPException
from pydantic import BaseModel
from generic_models import APIResponse, PaginatedResponse, Result

app = FastAPI(title="Generic Models Demo")

class User(BaseModel):
    id: int
    name: str
    email: str

# Simulated database
USERS = [User(id=i, name=f"User {i}", email=f"user{i}@example.com")
         for i in range(1, 101)]

@app.get("/users", response_model=APIResponse[PaginatedResponse[User]])
async def list_users(
    page: int = Query(1, ge=1),
    per_page: int = Query(10, ge=1, le=50)
):
    paginated = PaginatedResponse.from_sequence(USERS, page, per_page)
    return APIResponse.ok(paginated)

@app.get("/users/{user_id}", response_model=APIResponse[User])
async def get_user(user_id: int):
    user = next((u for u in USERS if u.id == user_id), None)
    if not user:
        return APIResponse.error(f"User {user_id} not found")
    return APIResponse.ok(user)
  2. Verify OpenAPI schema at /docs shows correct nested types.

Checkpoint: FastAPI endpoints work with generic response models.

Testing Strategy

Unit Tests for Generic Models

# tests/test_api_response.py
import pytest
from pydantic import BaseModel
from generic_models import APIResponse

class User(BaseModel):
    id: int
    name: str

def test_api_response_ok():
    """Test successful response creation."""
    user = User(id=1, name="John")
    response = APIResponse[User].ok(user, message="User found")

    assert response.success is True
    assert response.data == user
    assert response.message == "User found"
    assert response.errors is None

def test_api_response_error():
    """Test error response creation."""
    response = APIResponse[User].error(
        "Not found",
        errors=["User does not exist"]
    )

    assert response.success is False
    assert response.data is None
    assert response.message == "Not found"
    assert response.errors == ["User does not exist"]

def test_api_response_serialization():
    """Test JSON serialization."""
    user = User(id=1, name="John")
    response = APIResponse[User].ok(user)

    data = response.model_dump()
    assert data["success"] is True
    assert data["data"]["id"] == 1
    assert data["data"]["name"] == "John"

def test_api_response_nested_generic():
    """Test with nested generic types."""
    from generic_models import PaginatedResponse

    users = [User(id=i, name=f"User {i}") for i in range(1, 4)]
    paginated = PaginatedResponse[User](items=users, total=3, page=1, per_page=10)
    response = APIResponse[PaginatedResponse[User]].ok(paginated)

    assert response.success is True
    assert len(response.data.items) == 3
    assert response.data.total == 3

Tests for PaginatedResponse

# tests/test_paginated.py
import pytest
from pydantic import BaseModel
from generic_models import PaginatedResponse

class Item(BaseModel):
    id: int
    name: str

def test_pagination_properties():
    """Test computed properties."""
    paginated = PaginatedResponse[Item](
        items=[Item(id=1, name="Test")],
        total=100,
        page=5,
        per_page=10
    )

    assert paginated.pages == 10
    assert paginated.has_next is True
    assert paginated.has_prev is True

def test_pagination_first_page():
    """Test first page has no prev."""
    paginated = PaginatedResponse[Item](
        items=[],
        total=100,
        page=1,
        per_page=10
    )

    assert paginated.has_prev is False
    assert paginated.has_next is True

def test_pagination_last_page():
    """Test last page has no next."""
    paginated = PaginatedResponse[Item](
        items=[],
        total=100,
        page=10,
        per_page=10
    )

    assert paginated.has_prev is True
    assert paginated.has_next is False

def test_from_sequence():
    """Test creating paginated response from sequence."""
    items = [Item(id=i, name=f"Item {i}") for i in range(1, 101)]

    paginated = PaginatedResponse.from_sequence(items, page=2, per_page=10)

    assert len(paginated.items) == 10
    assert paginated.items[0].id == 11  # Second page starts at 11
    assert paginated.total == 100
    assert paginated.pages == 10

def test_from_sequence_partial_page():
    """Test when last page is partial."""
    items = [Item(id=i, name=f"Item {i}") for i in range(1, 26)]  # 25 items

    paginated = PaginatedResponse.from_sequence(items, page=3, per_page=10)

    assert len(paginated.items) == 5  # Only 5 items on last page
    assert paginated.total == 25
    assert paginated.pages == 3

Tests for Result Type

# tests/test_result.py
import pytest
from generic_models import Result

def test_result_ok():
    """Test success result."""
    result = Result[int, str].ok(42)

    assert result.is_ok() is True
    assert result.is_err() is False
    assert result.unwrap() == 42

def test_result_err():
    """Test error result."""
    result = Result[int, str].err("Something went wrong")

    assert result.is_ok() is False
    assert result.is_err() is True
    assert result.unwrap_err() == "Something went wrong"

def test_result_unwrap_or():
    """Test unwrap_or with default."""
    ok_result = Result[int, str].ok(42)
    err_result = Result[int, str].err("error")

    assert ok_result.unwrap_or(0) == 42
    assert err_result.unwrap_or(0) == 0

def test_result_map():
    """Test mapping over success value."""
    result = Result[int, str].ok(5)
    mapped = result.map(lambda x: x * 2)

    assert mapped.is_ok()
    assert mapped.unwrap() == 10

def test_result_map_on_error():
    """Test map does nothing on error."""
    result = Result[int, str].err("error")
    mapped = result.map(lambda x: x * 2)

    assert mapped.is_err()
    assert mapped.unwrap_err() == "error"

def test_result_map_err():
    """Test mapping over error value."""
    result = Result[int, str].err("error")
    mapped = result.map_err(lambda e: f"Wrapped: {e}")

    assert mapped.is_err()
    assert mapped.unwrap_err() == "Wrapped: error"

Integration Tests with FastAPI

# tests/test_fastapi_integration.py
import pytest
from fastapi.testclient import TestClient
from examples.fastapi_app import app

client = TestClient(app)

def test_list_users_returns_paginated():
    """Test that list users returns paginated response."""
    response = client.get("/users?page=1&per_page=5")

    assert response.status_code == 200
    data = response.json()

    assert data["success"] is True
    assert len(data["data"]["items"]) == 5
    assert data["data"]["total"] == 100
    assert data["data"]["page"] == 1
    assert data["data"]["has_next"] is True

def test_list_users_last_page():
    """Test last page pagination."""
    response = client.get("/users?page=10&per_page=10")

    data = response.json()
    assert data["data"]["has_next"] is False
    assert data["data"]["has_prev"] is True

def test_get_user_success():
    """Test getting existing user."""
    response = client.get("/users/1")

    assert response.status_code == 200
    data = response.json()

    assert data["success"] is True
    assert data["data"]["id"] == 1

def test_get_user_not_found():
    """Test getting non-existent user."""
    response = client.get("/users/999")

    assert response.status_code == 200  # We return 200 with error in body
    data = response.json()

    assert data["success"] is False
    assert "not found" in data["message"].lower()

def test_openapi_schema():
    """Test that OpenAPI schema is generated correctly."""
    response = client.get("/openapi.json")

    assert response.status_code == 200
    schema = response.json()

    # Check that generic types are in schemas
    schemas = schema["components"]["schemas"]
    assert any("APIResponse" in name for name in schemas)
    assert any("PaginatedResponse" in name for name in schemas)

Common Pitfalls and Debugging

Pitfall 1: Forgetting to Inherit from Both BaseModel and Generic

Problem: Model doesn’t behave as generic.

# WRONG - missing Generic[T]
class Wrapper(BaseModel):
    data: T  # T is not a type parameter!

# CORRECT
class Wrapper(BaseModel, Generic[T]):
    data: T

Symptom: T is treated as a forward reference to a class named “T”.

Solution: Always inherit from both BaseModel and Generic[T].

Pitfall 2: TypeVar Scope Issues

Problem: TypeVar defined in wrong scope.

# WRONG - T defined inside class
class Container(BaseModel, Generic[T]):  # T not defined yet!
    T = TypeVar('T')  # This is a class attribute, not a type parameter
    data: T

# CORRECT - T defined at module level
T = TypeVar('T')

class Container(BaseModel, Generic[T]):
    data: T

Solution: Define TypeVars at module level before using them.

Pitfall 3: Mutable Default Arguments in Generic Classes

Problem: Shared mutable defaults across instances.

# WRONG - mutable default shared
class Response(BaseModel, Generic[T]):
    items: list[T] = []  # Same list for all instances!

# CORRECT - use Field with default_factory
class Response(BaseModel, Generic[T]):
    items: list[T] = Field(default_factory=list)

Symptom: Data mysteriously appears in “empty” instances.

Pitfall 4: Type Parameter Not Being Validated

Problem: Pydantic doesn’t validate the type parameter at runtime.

T = TypeVar('T')

class Box(BaseModel, Generic[T]):
    value: T

# This works even though we said Box[int]!
box: Box[int] = Box(value="not an int")  # No validation error!

Why: Generic type parameters are hints for static analysis. Pydantic validates when you use a concrete type.

Solution: Use parameterized types for validation:

IntBox = Box[int]
int_box = IntBox(value="not an int")  # Now raises ValidationError!

Pitfall 5: Computed Fields with Generics

Problem: Using @property instead of @computed_field.

# WRONG - property not included in model_dump()
class Response(BaseModel, Generic[T]):
    items: list[T]

    @property
    def count(self) -> int:
        return len(self.items)

response.model_dump()  # count not included!

# CORRECT - use computed_field
from pydantic import computed_field

class Response(BaseModel, Generic[T]):
    items: list[T]

    @computed_field
    @property
    def count(self) -> int:
        return len(self.items)

response.model_dump()  # {"items": [...], "count": 5}

Pitfall 6: Circular Type References

Problem: Generic models that reference each other.

# WRONG - NameError: Container not defined
T = TypeVar('T')

class Item(BaseModel, Generic[T]):
    container: Container[T]  # Container not defined yet!

class Container(BaseModel, Generic[T]):
    items: list[Item[T]]

# CORRECT - use forward references
class Item(BaseModel, Generic[T]):
    container: "Container[T]"  # String forward reference

class Container(BaseModel, Generic[T]):
    items: list["Item[T]"]

# Rebuild models after both are defined
Item.model_rebuild()
Container.model_rebuild()

Debugging Tips

  1. Check type at runtime: 
    print(type(response))
print(response.__class__.__bases__)
print(getattr(response, '__pydantic_generic_metadata__', None))
  2. Inspect generic parameters: 
    from typing import get_args, get_origin

alias = APIResponse[User]
print(get_origin(alias))  # <class 'APIResponse'>
print(get_args(alias))    # (<class 'User'>,)
  3. View generated schema: 
    print(APIResponse[User].model_json_schema())

Extensions and Challenges

Extension 1: Async Result Type

Create an async-aware Result type that can hold coroutines:

from typing import Awaitable, Callable

class AsyncResult(Result[T, E]):
    """Result type for async operations."""

    async def map_async(
        self,
        fn: Callable[[T], Awaitable[U]]
    ) -> "AsyncResult[U, E]":
        if self.is_ok():
            new_value = await fn(self.value)
            return AsyncResult[U, E].ok(new_value)
        return AsyncResult[U, E].err(self.error)

    @classmethod
    async def from_coroutine(
        cls,
        coro: Awaitable[T]
    ) -> "AsyncResult[T, Exception]":
        """Wrap a coroutine in a Result."""
        try:
            value = await coro
            return cls.ok(value)
        except Exception as e:
            return cls.err(e)

Extension 2: Cursor-Based Pagination

Implement cursor pagination for large datasets:

class CursorPage(BaseModel, Generic[T]):
    """Cursor-based pagination for infinite scroll."""
    items: list[T]
    next_cursor: str | None = None
    prev_cursor: str | None = None
    has_more: bool = False

    @classmethod
    def create(
        cls,
        items: list[T],
        cursor_field: str = "id",
        limit: int = 10
    ) -> "CursorPage[T]":
        """Create page with cursor based on field."""
        if not items:
            return cls(items=[], has_more=False)

        # Encode last item's cursor field as next cursor
        last_item = items[-1]
        next_cursor = encode_cursor(getattr(last_item, cursor_field))

        return cls(
            items=items,
            next_cursor=next_cursor if len(items) == limit else None,
            has_more=len(items) == limit
        )

Extension 3: Type-Safe Factory Pattern

Create a factory that produces correctly typed generic models:

class ModelFactory(Generic[T]):
    """Factory for creating instances of generic models."""

    def __init__(self, model_type: Type[T]):
        self.model_type = model_type

    def create_response(self, data: T) -> APIResponse[T]:
        return APIResponse[T].ok(data)

    def create_page(self, items: list[T], **kwargs) -> PaginatedResponse[T]:
        return PaginatedResponse[T](items=items, **kwargs)

    def create_result(self, value: T) -> Result[T, Exception]:
        return Result[T, Exception].ok(value)

# Usage
user_factory = ModelFactory(User)
response = user_factory.create_response(user)  # APIResponse[User]

Extension 4: Validation Context

Add context to generic models for conditional validation:

from pydantic import ValidationInfo

class ContextualResponse(BaseModel, Generic[T]):
    """Response with context-aware validation."""
    data: T
    context: dict = Field(default_factory=dict)

    @model_validator(mode='after')
    def validate_with_context(self, info: ValidationInfo):
        """Apply context-specific validation."""
        if info.context and info.context.get('strict'):
            # Additional validation in strict mode
            if self.data is None:
                raise ValueError("Data required in strict mode")
        return self

Extension 5: Streaming Response Support

Create generic models for streaming responses:

from typing import AsyncIterator

class StreamingResponse(BaseModel, Generic[T]):
    """Wrapper for streaming responses."""
    started_at: datetime
    total_expected: int | None = None

    class Config:
        arbitrary_types_allowed = True

    async def stream(
        self,
        source: AsyncIterator[T]
    ) -> AsyncIterator[StreamChunk[T]]:
        """Stream items with metadata."""
        count = 0
        async for item in source:
            count += 1
            yield StreamChunk(
                item=item,
                sequence=count,
                is_last=False
            )
        yield StreamChunk(item=None, sequence=count, is_last=True)

class StreamChunk(BaseModel, Generic[T]):
    item: T | None
    sequence: int
    is_last: bool

Real-World Connections

Where Generic Models Appear

  1. FastAPI Response Models
    • Most FastAPI applications use generic response wrappers
    • Standard patterns: APIResponse[T], PagedResult[T]
  2. GraphQL Resolvers
    • Relay-style connections use generics: Connection[T], Edge[T]
    • Strawberry GraphQL supports generic types
  3. Database Libraries
    • SQLAlchemy 2.0 uses generics: Mapped[T]
    • SQLModel combines Pydantic generics with ORM
  4. Functional Programming Libraries
    • returns library: Result, Maybe, IO types
    • Railway-oriented programming patterns
  5. Message Queues
    • Kafka/RabbitMQ wrappers: Message[T], Event[T]
    • Type-safe message serialization

Industry Examples

  • Stripe API: Uses envelope patterns (data, has_more, object)
  • GitHub GraphQL: Uses connection/edge pattern for pagination
  • AWS SDK: Uses Result-like types for operation outcomes
  • OpenAI API: Structured output uses Pydantic generics

Production Considerations

  1. OpenAPI Schema Generation
    • Generic types generate unique schema names (e.g., APIResponse_User_)
    • Consider using schema_extra for cleaner names
  2. Performance
    • Generic model creation has minimal overhead
    • Use model_construct() for trusted data to skip validation
  3. Caching
    • Pydantic caches parameterized types
    • Wrapper[User] creates type once, reuses thereafter
  4. Documentation
    • Generic types appear correctly in Swagger/ReDoc
    • Add docstrings for better auto-documentation

Self-Assessment Checklist

Core Understanding

  • Can I explain what a TypeVar is and when to use bounds?
  • Can I describe the difference between covariance and contravariance?
  • Can I explain how Pydantic handles generic types at runtime?
  • Can I use get_origin and get_args for type introspection?

Implementation Skills

  • Can I create a generic BaseModel subclass?
  • Can I implement factory methods that preserve type information?
  • Can I compose multiple generic types (nested generics)?
  • Can I add computed properties to generic models?

Integration Knowledge

  • Can I use generic models as FastAPI response types?
  • Can I verify OpenAPI schema generation is correct?
  • Can I write tests for generic model behavior?
  • Can I debug type-related issues in generic code?

Mastery Indicators

  • Library works correctly with any Pydantic model type
  • FastAPI generates correct OpenAPI schemas
  • All edge cases are tested (empty lists, null values, etc.)
  • Code is well-documented and type-checker friendly

Resources

Documentation

Books

  • “Fluent Python” by Luciano Ramalho - Chapter 15 on Type Hints and Generics
  • “Robust Python” by Patrick Viafore - Chapter 7 on Generics

Articles and Tutorials