Skip to content

Version with paths and numbers instead of headers and dates

Cadwyn uses version headers with ISO dates by default for versioning. However, you can use any strings instead of ISO dates and/or you can use path version prefixes instead of version headers. Here's our quickstart tutorial example but using version numbers and path prefixes:

Feel free to mix and match the API version formats and version locations as you see fit. But beware that Cadwyn does not support version waterfalling for arbitrary strings as versions.

import uuid

from cadwyn import VersionedAPIRouter
from cadwyn.applications import Cadwyn
from cadwyn.structure import (
    RequestInfo,
    ResponseInfo,
    Version,
    VersionBundle,
    VersionChange,
    convert_request_to_next_version_for,
    convert_response_to_previous_version_for,
    endpoint,
    schema,
)
from cadwyn.structure.versions import HeadVersion
from pydantic import BaseModel, Field


class BaseUser(BaseModel):
    pass


class UserCreateRequest(BaseUser):
    default_address: str
    addresses_to_create: list[str] = Field(default_factory=list)


class UserResource(BaseUser):
    id: uuid.UUID


class UserAddressResource(BaseModel):
    id: uuid.UUID
    value: str


class UserAddressResourceList(BaseModel):
    data: list[UserAddressResource]


class ReplaceUserAddressWithListOfAddresses(VersionChange):
    """Users can now store multiple addresses instead of a single address so they can choose among delivery locations."""

    instructions_to_migrate_to_previous_version = (
        schema(BaseUser).field("addresses").didnt_exist,
        schema(BaseUser).field("address").existed_as(type=str, info=Field()),
    )

    @convert_request_to_next_version_for(UserCreateRequest)
    def change_address_to_multiple_items(request: RequestInfo) -> None:
        request.body["addresses"] = [request.body.pop("address")]

    @convert_response_to_previous_version_for(UserResource)
    def change_addresses_to_single_item(response: ResponseInfo) -> None:
        # Need to assert addresses length somewhere in business logic
        response.body["address"] = response.body["addresses"][0]


class MoveUserAddressesToSubresource(VersionChange):
    """User addresses are now separate resources with stable IDs, allowing clients to retrieve each address independently."""

    instructions_to_migrate_to_previous_version = (
        schema(BaseUser)
        .field("addresses")
        .existed_as(type=list[str], info=Field()),
        schema(UserCreateRequest).field("default_address").didnt_exist,
        endpoint(
            "/{api_version}/users/{user_id}/addresses", ["GET"]
        ).didnt_exist,
    )

    @convert_request_to_next_version_for(UserCreateRequest)
    def change_addresses_to_default_address(request: RequestInfo):
        request.body["default_address"] = request.body["addresses"].pop(0)
        # Save data to still be able to keep the old behavior of creating addresses
        request.body["addresses_to_create"] = request.body.pop("addresses")

    @convert_response_to_previous_version_for(UserResource)
    def change_addresses_to_list(response: ResponseInfo) -> None:
        response.body["addresses"] = [
            id["value"] for id in response.body["_prefetched_addresses"]
        ]


class RemoveAddressesToCreateFromLatestUserSchema(VersionChange):
    """Stop accepting 'addresses_to_create' when creating users.

    Additional addresses are now managed as separate resources.
    """

    instructions_to_migrate_to_previous_version = (
        schema(UserCreateRequest).field("addresses_to_create").didnt_exist,
    )


version_bundle = VersionBundle(
    HeadVersion(RemoveAddressesToCreateFromLatestUserSchema),
    Version("v10", MoveUserAddressesToSubresource),
    Version("v9", ReplaceUserAddressWithListOfAddresses),
    Version("v8"),
)


router = VersionedAPIRouter(tags=["Users"])
database_parody = {}


@router.post("/{api_version}/users", response_model=UserResource)
async def create_user(user: UserCreateRequest):
    id_ = uuid.uuid4()
    database_parody[id_] = {"id": id_}
    addresses = create_user_addresses(
        id_, [user.default_address, *user.addresses_to_create]
    )
    return database_parody[id_] | {"_prefetched_addresses": addresses}


@router.get("/{api_version}/users/{user_id}", response_model=UserResource)
async def get_user(user_id: uuid.UUID):
    return {
        "id": user_id,
        "_prefetched_addresses": (await get_user_addresses(user_id))["data"],
    }


def create_user_addresses(user_id: uuid.UUID, addresses: list[str]):
    database_parody[f"addr_{user_id}"] = [
        {"id": uuid.uuid4(), "value": address} for address in addresses
    ]
    return database_parody[f"addr_{user_id}"]


@router.get(
    "/{api_version}/users/{user_id}/addresses",
    response_model=UserAddressResourceList,
)
async def get_user_addresses(user_id: uuid.UUID):
    return {"data": database_parody[f"addr_{user_id}"]}


app = Cadwyn(
    versions=version_bundle,
    api_version_parameter_name="api_version",
    api_version_format="string",
    api_version_location="path",
)
app.generate_and_include_versioned_routers(router)