Skip to content

feat: extend type ergonomics to response types #106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bokelley merged 2 commits into from
Dec 21, 2025
+511 −22
Merged

feat: extend type ergonomics to response types #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7df2abf
feat: extend type ergonomics to response types
bokelley Dec 21, 2025
4c9f089
refactor: improve test imports and documentation
bokelley Dec 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions CLAUDE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,26 @@ FormatId = str
PackageRequest = dict[str, Any]
```

**Import Architecture for Generated Types**
The type system has a strict layering to prevent brittleness:

```
generated_poc/*.py (internal, auto-generated from schemas)
↓
_generated.py (internal consolidation)
↓
stable.py + aliases.py + _ergonomic.py (public API / internal infrastructure)
↓
__init__.py (user-facing exports)
```

Only these modules may import from `generated_poc/` or `_generated.py`:
- `stable.py`: Re-exports base types with clean names
- `aliases.py`: Creates semantic aliases for numbered discriminated union types
- `_ergonomic.py`: Applies BeforeValidator coercion for type ergonomics

All other source code should import from `adcp.types` (the public API).

**Type Checking Best Practices**
- Use `TYPE_CHECKING` for optional dependencies to avoid runtime import errors
- Use `cast()` for JSON deserialization to satisfy mypy's `no-any-return` checks
Expand Down
110 changes: 88 additions & 22 deletions scripts/generate_ergonomic_coercion.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,15 @@
"CreateMediaBuyRequest",
]

# Response types to analyze for coercion
RESPONSE_TYPES_TO_ANALYZE = [
"GetProductsResponse",
"ListCreativesResponse",
"ListCreativeFormatsResponse",
"CreateMediaBuyResponse1",
"GetMediaBuyDeliveryResponse",
]

# Nested types that also need coercion
NESTED_TYPES_TO_ANALYZE = [
("Sort", "media_buy.list_creatives_request"),
Expand All @@ -45,9 +54,18 @@

# Types that should get subclass_list coercion (for list variance)
SUBCLASS_LIST_TYPES = {
# Request list types
"CreativeAsset",
"CreativeAssignment",
"PackageRequest",
# Response list types
"Product",
"Creative",
"Format",
"Package",
"MediaBuyDelivery",
"Error",
"CreativeAgent",
}


Expand Down Expand Up @@ -180,14 +198,23 @@ def get_import_path(cls) -> str:
def generate_code() -> str:
"""Generate the _ergonomic.py module content."""
# Import all the types we need to analyze
from adcp.types.generated_poc.core.context import ContextObject
from adcp.types.generated_poc.core.creative_asset import CreativeAsset
from adcp.types.generated_poc.core.creative_assignment import CreativeAssignment
from adcp.types.generated_poc.core.ext import ExtensionObject
from adcp.types.generated_poc.media_buy.create_media_buy_request import CreateMediaBuyRequest
from adcp.types.generated_poc.media_buy.create_media_buy_response import CreateMediaBuyResponse1
from adcp.types.generated_poc.media_buy.get_media_buy_delivery_response import (
GetMediaBuyDeliveryResponse,
)
from adcp.types.generated_poc.media_buy.get_products_request import GetProductsRequest
from adcp.types.generated_poc.media_buy.list_creative_formats_request import ListCreativeFormatsRequest

# Response types
from adcp.types.generated_poc.media_buy.get_products_response import GetProductsResponse
from adcp.types.generated_poc.media_buy.list_creative_formats_request import (
ListCreativeFormatsRequest,
)
from adcp.types.generated_poc.media_buy.list_creative_formats_response import (
ListCreativeFormatsResponse,
)
from adcp.types.generated_poc.media_buy.list_creatives_request import ListCreativesRequest, Sort
from adcp.types.generated_poc.media_buy.list_creatives_response import ListCreativesResponse
from adcp.types.generated_poc.media_buy.package_request import PackageRequest
from adcp.types.generated_poc.media_buy.update_media_buy_request import Packages, Packages1

Expand All @@ -200,6 +227,14 @@ def generate_code() -> str:
"CreateMediaBuyRequest": CreateMediaBuyRequest,
}

response_classes = {
"GetProductsResponse": GetProductsResponse,
"ListCreativesResponse": ListCreativesResponse,
"ListCreativeFormatsResponse": ListCreativeFormatsResponse,
"CreateMediaBuyResponse1": CreateMediaBuyResponse1,
"GetMediaBuyDeliveryResponse": GetMediaBuyDeliveryResponse,
}

nested_classes = {
"Sort": Sort,
"Packages": Packages,
Expand All @@ -210,7 +245,7 @@ def generate_code() -> str:
all_coercions = {}
all_imports = set()

for name, cls in {**request_classes, **nested_classes}.items():
for name, cls in {**request_classes, **response_classes, **nested_classes}.items():
coercions = analyze_model(cls)
if coercions:
all_coercions[name] = (cls, coercions)
Expand Down Expand Up @@ -238,6 +273,10 @@ def generate_code() -> str:
core_imports.append(("ExtensionObject", "core.ext"))
core_imports.append(("CreativeAsset", "core.creative_asset"))
core_imports.append(("CreativeAssignment", "core.creative_assignment"))
core_imports.append(("Product", "core.product"))
core_imports.append(("Format", "core.format"))
core_imports.append(("Package", "core.package"))
core_imports.append(("Error", "core.error"))

# Deduplicate
enum_imports = sorted(set(enum_imports))
Expand Down Expand Up @@ -317,6 +356,26 @@ def generate_code() -> str:
lines.append(' Packages1,')
lines.append(')')

# Add response type imports
lines.append('# Response types')
lines.append('from adcp.types.generated_poc.media_buy.create_media_buy_response import (')
lines.append(' CreateMediaBuyResponse1,')
lines.append(')')
lines.append('from adcp.types.generated_poc.media_buy.get_media_buy_delivery_response import (')
lines.append(' GetMediaBuyDeliveryResponse,')
lines.append(' MediaBuyDelivery,')
lines.append(' NotificationType,')
lines.append(')')
lines.append('from adcp.types.generated_poc.media_buy.get_products_response import GetProductsResponse')
lines.append('from adcp.types.generated_poc.media_buy.list_creative_formats_response import (')
lines.append(' CreativeAgent,')
lines.append(' ListCreativeFormatsResponse,')
lines.append(')')
lines.append('from adcp.types.generated_poc.media_buy.list_creatives_response import (')
lines.append(' Creative,')
lines.append(' ListCreativesResponse,')
lines.append(')')

lines.append('')
lines.append('')
lines.append('def _apply_coercion() -> None:')
Expand All @@ -329,6 +388,7 @@ def generate_code() -> str:
# Generate coercion code for each type
# Process in a specific order for readability
type_order = [
# Request types
"ListCreativeFormatsRequest",
"ListCreativesRequest",
"Sort",
Expand All @@ -337,6 +397,12 @@ def generate_code() -> str:
"CreateMediaBuyRequest",
"Packages",
"Packages1",
# Response types
"GetProductsResponse",
"ListCreativesResponse",
"ListCreativeFormatsResponse",
"CreateMediaBuyResponse1",
"GetMediaBuyDeliveryResponse",
]

for type_name in type_order:
Expand Down Expand Up @@ -368,47 +434,47 @@ def generate_code() -> str:
field = c["field"]
if c["type"] == "enum":
target = c["target_class"].__name__
lines.append(f' _patch_field_annotation(')
lines.append(' _patch_field_annotation(')
lines.append(f' {type_name},')
lines.append(f' "{field}",')
lines.append(f' Annotated[{target} | None, BeforeValidator(coerce_to_enum({target}))],')
lines.append(f' )')
lines.append(' )')
elif c["type"] == "enum_list":
target = c["target_class"].__name__
lines.append(f' _patch_field_annotation(')
lines.append(' _patch_field_annotation(')
lines.append(f' {type_name},')
lines.append(f' "{field}",')
lines.append(f' Annotated[')
lines.append(' Annotated[')
lines.append(f' list[{target}] | None,')
lines.append(f' BeforeValidator(coerce_to_enum_list({target})),')
lines.append(f' ],')
lines.append(f' )')
lines.append(' ],')
lines.append(' )')
elif c["type"] == "context":
lines.append(f' _patch_field_annotation(')
lines.append(' _patch_field_annotation(')
lines.append(f' {type_name},')
lines.append(f' "{field}",')
lines.append(f' Annotated[ContextObject | None, BeforeValidator(coerce_to_model(ContextObject))],')
lines.append(f' )')
lines.append(' Annotated[ContextObject | None, BeforeValidator(coerce_to_model(ContextObject))],')
lines.append(' )')
elif c["type"] == "ext":
lines.append(f' _patch_field_annotation(')
lines.append(' _patch_field_annotation(')
lines.append(f' {type_name},')
lines.append(f' "{field}",')
lines.append(f' Annotated[ExtensionObject | None, BeforeValidator(coerce_to_model(ExtensionObject))],')
lines.append(f' )')
lines.append(' Annotated[ExtensionObject | None, BeforeValidator(coerce_to_model(ExtensionObject))],')
lines.append(' )')
elif c["type"] == "subclass_list":
target = c["target_class"].__name__
# Check if the field is required (no | None)
field_info = cls.model_fields[field]
is_optional = "None" in str(field_info.annotation)
type_str = f'list[{target}] | None' if is_optional else f'list[{target}]'
lines.append(f' _patch_field_annotation(')
lines.append(' _patch_field_annotation(')
lines.append(f' {type_name},')
lines.append(f' "{field}",')
lines.append(f' Annotated[')
lines.append(' Annotated[')
lines.append(f' {type_str},')
lines.append(f' BeforeValidator(coerce_subclass_list({target})),')
lines.append(f' ],')
lines.append(f' )')
lines.append(' ],')
lines.append(' )')

lines.append(f' {type_name}.model_rebuild(force=True)')
lines.append('')
Expand Down
Loading