fix: duplicate union type declarations for Stainless codegen

Extract duplicate union types to shared schema references and remove
duplicate references within unions to fix Stainless duplicate declaration
warnings.

Fixes:
https://www.stainless.com/docs/reference/diagnostics#Python/DuplicateDeclaration

Signed-off-by: Sébastien Han <seb@redhat.com>
This commit is contained in:
Sébastien Han 2025-11-13 09:58:16 +01:00
parent 09280301de
commit 01f441b3ac
No known key found for this signature in database
6 changed files with 476 additions and 484 deletions

View file

@ -1470,6 +1470,139 @@ def _remove_request_bodies_from_get_endpoints(openapi_schema: dict[str, Any]) ->
return openapi_schema
def _extract_duplicate_union_types(openapi_schema: dict[str, Any]) -> dict[str, Any]:
"""
Extract duplicate union types to shared schema references.
Stainless generates type names from union types based on their context, which can cause
duplicate names when the same union appears in different places. This function extracts
these duplicate unions to shared schema definitions and replaces inline definitions with
references to them.
According to Stainless docs, when duplicate types are detected, they should be extracted
to the same ref and declared as a model. This ensures Stainless generates consistent
type names regardless of where the union is referenced.
Fixes: https://www.stainless.com/docs/reference/diagnostics#Python/DuplicateDeclaration
"""
import copy
if "components" not in openapi_schema or "schemas" not in openapi_schema["components"]:
return openapi_schema
schemas = openapi_schema["components"]["schemas"]
# Extract the Output union type (used in OpenAIResponseObjectWithInput-Output and ListOpenAIResponseInputItem)
output_union_schema_name = "OpenAIResponseMessageOutputUnion"
output_union_title = None
# Get the union type from OpenAIResponseObjectWithInput-Output.input.items.anyOf
if "OpenAIResponseObjectWithInput-Output" in schemas:
schema = schemas["OpenAIResponseObjectWithInput-Output"]
if isinstance(schema, dict) and "properties" in schema:
input_prop = schema["properties"].get("input")
if isinstance(input_prop, dict) and "items" in input_prop:
items = input_prop["items"]
if isinstance(items, dict) and "anyOf" in items:
# Extract the union schema with deep copy
output_union_schema = copy.deepcopy(items["anyOf"])
output_union_title = items.get("title", "OpenAIResponseMessageOutputUnion")
# Collect all refs from the oneOf to detect duplicates
refs_in_oneof = set()
for item in output_union_schema:
if isinstance(item, dict) and "oneOf" in item:
oneof = item["oneOf"]
if isinstance(oneof, list):
for variant in oneof:
if isinstance(variant, dict) and "$ref" in variant:
refs_in_oneof.add(variant["$ref"])
item["x-stainless-naming"] = "OpenAIResponseMessageOutputOneOf"
# Remove duplicate refs from anyOf that are already in oneOf
deduplicated_schema = []
for item in output_union_schema:
if isinstance(item, dict) and "$ref" in item:
if item["$ref"] not in refs_in_oneof:
deduplicated_schema.append(item)
else:
deduplicated_schema.append(item)
output_union_schema = deduplicated_schema
# Create the shared schema with x-stainless-naming to ensure consistent naming
if output_union_schema_name not in schemas:
schemas[output_union_schema_name] = {
"anyOf": output_union_schema,
"title": output_union_title,
"x-stainless-naming": output_union_schema_name,
}
# Replace with reference
input_prop["items"] = {"$ref": f"#/components/schemas/{output_union_schema_name}"}
# Replace the same union in ListOpenAIResponseInputItem.data.items.anyOf
if "ListOpenAIResponseInputItem" in schemas and output_union_schema_name in schemas:
schema = schemas["ListOpenAIResponseInputItem"]
if isinstance(schema, dict) and "properties" in schema:
data_prop = schema["properties"].get("data")
if isinstance(data_prop, dict) and "items" in data_prop:
items = data_prop["items"]
if isinstance(items, dict) and "anyOf" in items:
# Replace with reference
data_prop["items"] = {"$ref": f"#/components/schemas/{output_union_schema_name}"}
# Extract the Input union type (used in _responses_Request.input.anyOf[1].items.anyOf)
input_union_schema_name = "OpenAIResponseMessageInputUnion"
if "_responses_Request" in schemas:
schema = schemas["_responses_Request"]
if isinstance(schema, dict) and "properties" in schema:
input_prop = schema["properties"].get("input")
if isinstance(input_prop, dict) and "anyOf" in input_prop:
any_of = input_prop["anyOf"]
if isinstance(any_of, list) and len(any_of) > 1:
# Check the second item (index 1) which should be the array type
second_item = any_of[1]
if isinstance(second_item, dict) and "items" in second_item:
items = second_item["items"]
if isinstance(items, dict) and "anyOf" in items:
# Extract the union schema with deep copy
input_union_schema = copy.deepcopy(items["anyOf"])
input_union_title = items.get("title", "OpenAIResponseMessageInputUnion")
# Collect all refs from the oneOf to detect duplicates
refs_in_oneof = set()
for item in input_union_schema:
if isinstance(item, dict) and "oneOf" in item:
oneof = item["oneOf"]
if isinstance(oneof, list):
for variant in oneof:
if isinstance(variant, dict) and "$ref" in variant:
refs_in_oneof.add(variant["$ref"])
item["x-stainless-naming"] = "OpenAIResponseMessageInputOneOf"
# Remove duplicate refs from anyOf that are already in oneOf
deduplicated_schema = []
for item in input_union_schema:
if isinstance(item, dict) and "$ref" in item:
if item["$ref"] not in refs_in_oneof:
deduplicated_schema.append(item)
else:
deduplicated_schema.append(item)
input_union_schema = deduplicated_schema
# Create the shared schema with x-stainless-naming to ensure consistent naming
if input_union_schema_name not in schemas:
schemas[input_union_schema_name] = {
"anyOf": input_union_schema,
"title": input_union_title,
"x-stainless-naming": input_union_schema_name,
}
# Replace with reference
second_item["items"] = {"$ref": f"#/components/schemas/{input_union_schema_name}"}
return openapi_schema
def _convert_multiline_strings_to_literal(obj: Any) -> Any:
"""Recursively convert multi-line strings to LiteralScalarString for YAML block scalar formatting."""
try:
@ -1854,6 +1987,9 @@ def generate_openapi_spec(output_dir: str) -> dict[str, Any]:
# that FastAPI incorrectly added to GET endpoints are removed
openapi_schema = _remove_request_bodies_from_get_endpoints(openapi_schema)
# Extract duplicate union types to shared schema references
openapi_schema = _extract_duplicate_union_types(openapi_schema)
# Split into stable (v1 only), experimental (v1alpha + v1beta), deprecated, and combined (stainless) specs
# Each spec needs its own deep copy of the full schema to avoid cross-contamination
import copy
@ -1865,6 +2001,9 @@ def generate_openapi_spec(output_dir: str) -> dict[str, Any]:
deprecated_schema = _filter_deprecated_schema(copy.deepcopy(openapi_schema))
combined_schema = _filter_combined_schema(copy.deepcopy(openapi_schema))
# Apply duplicate union extraction to combined schema (used by Stainless)
combined_schema = _extract_duplicate_union_types(combined_schema)
base_description = (
"This is the specification of the Llama Stack that provides\n"
" a set of endpoints and their corresponding interfaces that are\n"