Schema - improvements

src/aiopenapi3/base.py

@@ -366,8 +366,6 @@ def __getstate__(self):
 
     def _get_identity(self, prefix="XLS", name=None):
         if self._identity is None:
-            if name is None:
-                name = self.title
             if name:
                 n = re.sub(r"\W", "_", name, flags=re.ASCII)
             else:

src/aiopenapi3/openapi.py

@@ -495,7 +495,9 @@ def is_schema(v: tuple[str, "SchemaType"]) -> bool:
                 for byid in map(lambda x: x.definitions, documents):
                     assert byid is not None and isinstance(byid, dict)
                     for name, schema in filter(is_schema, byid.items()):
-                        byname[schema._get_identity(name=name)] = schema
+                        n = schema._get_identity(name=name)
+                        # assert byname.get(n, None) in [None, schema]
+                        byname[n] = schema
 
                 # PathItems
                 for path, obj in (self.paths or dict()).items():
@@ -508,6 +510,7 @@ def is_schema(v: tuple[str, "SchemaType"]) -> bool:
                             if isinstance(response, (v20.paths.Response)):
                                 if isinstance(response.schema_, (v20.Schema, v31.Schema)):
                                     name = response.schema_._get_identity("PI", f"{path}.{m}.{r}")
+                                    # assert byname.get(name, None) in [None, response.schema_]
                                     byname[name] = response.schema_
                             else:
                                 raise TypeError(f"{type(response)} at {path}")
@@ -517,7 +520,9 @@ def is_schema(v: tuple[str, "SchemaType"]) -> bool:
                 assert byid is not None and isinstance(byid, dict)
                 for name, response in filter(is_schema, byid.items()):
                     assert response.schema_
-                    byname[response.schema_._get_identity(name=name)] = response.schema_
+                    n = response.schema_._get_identity(name=name)
+                    # assert byname.get(name, None) in [None, response.schema_]
+                    byname[n] = response.schema_
 
         elif isinstance(self._root, (v30.Root, v31.Root)):
             # Schema
@@ -528,7 +533,9 @@ def is_schema(v: tuple[str, "SchemaType"]) -> bool:
                 for byid in map(lambda x: x.schemas, components):
                     assert byid is not None and isinstance(byid, dict)
                     for name, schema in filter(is_schema, byid.items()):
-                        byname[schema._get_identity(name=name)] = schema
+                        n = schema._get_identity(name=name)
+                        # assert byname.get(n, None) in [None, schema]
+                        byname[n] = schema
 
             # PathItems
             for path, obj in (self.paths or dict()).items():
@@ -541,24 +548,28 @@ def is_schema(v: tuple[str, "SchemaType"]) -> bool:
                                 schema = parameter.schema_._target
                             else:
                                 schema = parameter.schema_
-                            assert schema is not None
+                            # assert schema is not None
                             name = schema._get_identity("I2", f"{path}.{m}.{parameter.name}")
+                            # assert byname.get(name, None) in [None, schema]
                             byname[name] = schema
                         else:
                             for key, mto in parameter.content.items():
                                 if isinstance(mto.schema_, ReferenceBase):
                                     schema = mto.schema_._target
                                 else:
                                     schema = mto.schema_
-                                assert schema is not None
+                                # assert schema is not None
                                 name = schema._get_identity("I2", f"{path}.{m}.{parameter.name}.{key}")
+                                # assert byname.get(name, None) in [None, schema]
                                 byname[name] = schema
 
                     if op.requestBody:
                         for mt, mto in op.requestBody.content.items():
                             if mto.schema_ is None:
                                 continue
-                            byname[mto.schema_._get_identity("B")] = mto.schema_
+                            n = mto.schema_._get_identity("B")
+                            # assert byname.get(n, None) in [None, getattr(mto.schema_, "_target", mto.schema_)]
+                            byname[n] = mto.schema_
 
                     for r, response in op.responses.items():
                         if isinstance(response, ReferenceBase):
@@ -569,6 +580,9 @@ def is_schema(v: tuple[str, "SchemaType"]) -> bool:
                                 if mto.schema_ is None:
                                     continue
                                 name = mto.schema_._get_identity("I2", f"{path}.{m}.{r}.{mt}")
+                                # assert (v := byname.get(name, None)) in [None, mto.schema_] or type(v) != type(
+                                #    mto.schema_
+                                # ), (name, v, mto.schema_)
                                 byname[name] = mto.schema_
                         else:
                             raise TypeError(f"{type(response)} at {path}")
@@ -581,7 +595,9 @@ def is_schema(v: tuple[str, "SchemaType"]) -> bool:
                         for mt, mto in response.content.items():
                             if mto.schema_ is None:
                                 continue
-                            byname[mto.schema_._get_identity("R")] = mto.schema_
+                            n = mto.schema_._get_identity("R")
+                            # assert byname.get(n, None) in [None, mto.schema_]
+                            byname[n] = mto.schema_
 
         byname = self.plugins.init.schemas(initialized=self._root, schemas=byname).schemas
         return byname
@@ -605,9 +621,21 @@ def _init_schema_types(self, only_required: bool) -> None:
         for i in todo | data:
             b = byid[i]
             name = b._get_identity("X")
-            types[name] = b.get_type()
-            for idx, j in enumerate(b._model_types):
-                types[f"{name}.c{idx}"] = j
+            t = b.get_type()
+            # assert (v := byname.get(name, None)) in [None, b], (name, b, v)
+            types[name] = t
+            for j in b._model_types:
+                types[j.__name__] = j
+
+        # as previous .get_type() may have created new models, we need to reindex
+        for name, schema in list(types.items()):
+            if not is_basemodel(schema):
+                continue
+            thes = byname.get(name, None)
+            if thes is not None:
+                for v in byid[id(thes)]._model_types:
+                    if v.__name__ not in types:
+                        types[v.__name__] = v
 
         # print(f"{len(types)}")
         for name, schema in types.items():
@@ -619,6 +647,7 @@ def _init_schema_types(self, only_required: bool) -> None:
                 thes = byname.get(name, None)
                 if thes is not None:
                     for v in byid[id(thes)]._model_types:
+                        assert v.__name__ in types, v.__name__
                         v.model_rebuild(_types_namespace={"__types": types})
             except Exception as e:
                 raise e

tests/conftest.py

@@ -561,3 +561,13 @@ def with_schema_oneOf_mixed():
 @pytest.fixture
 def with_schema_anyOf():
     yield _get_parsed_yaml("schema-anyOf.yaml")
+
+
+@pytest.fixture
+def with_schema_title_name_collision():
+    yield _get_parsed_yaml("schema-title-name-collision.yaml")
+
+
+@pytest.fixture
+def with_schema_discriminated_union_extends():
+    yield _get_parsed_yaml("schema-discriminated-union-extends.yaml")

tests/fixtures/schema-discriminated-union-extends.yaml

@@ -0,0 +1,42 @@
+openapi: 3.1.0
+
+info:
+  title: title collision
+  version: 2.2.7
+
+servers:
+  - url: /
+
+paths: {}
+
+
+components:
+  schemas:
+    A:
+      type: object
+      additionalProperties: false
+      properties:
+        a:
+          type: integer
+    B:
+      type: object
+      additionalProperties: false
+      properties:
+        b:
+          type: integer
+
+    AB:
+      type: object
+      additionalProperties: false
+      properties:
+        type:
+          type: string
+          const: ""
+      oneOf:
+        - $ref: "#/components/schemas/A"
+        - $ref: "#/components/schemas/B"
+      discriminator:
+        propertyName: type
+        mapping:
+          A: "#/components/schemas/A"
+          B: "#/components/schemas/B"

tests/fixtures/schema-title-name-collision.yaml

@@ -0,0 +1,29 @@
+openapi: 3.1.0
+
+info:
+  title: title collision
+  version: 2.2.7
+
+servers:
+  - url: /
+
+paths: {}
+
+
+components:
+  schemas:
+    B:
+      type: object
+      title: B
+      properties:
+        a:
+          title: A
+          type: string
+    A:
+      type: object
+      properties:
+        id:
+          type: integer
+
+    C:
+      $ref: '#/components/schemas/A'

tests/schema_test.py

@@ -50,7 +50,8 @@ def test_schema_without_properties(httpx_mock):
     assert result.example == "it worked"
 
     # the schema without properties did get its own named type defined
-    assert type(result.no_properties).__name__ == "has_no_properties"
+    # assert type(result.no_properties).__name__ == "has_no_properties"
+
     # and it has no fields
     assert len(type(result.no_properties).model_fields) == 0
 
@@ -790,3 +791,14 @@ def test_schema_date_types(with_schema_date_types):
     v = String.model_validate(str(ts))
     assert isinstance(v.root, datetime)
     assert v.model_dump_json()[1:-1] == now.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+
+def test_schema_title_name_collision(with_schema_title_name_collision):
+    api = OpenAPI("/", with_schema_title_name_collision)
+    C = api.components.schemas["C"].get_type()
+    C.model_validate({"a": 1})
+
+
+def test_schema_discriminated_union_extends(with_schema_discriminated_union_extends):
+    # AssertionError: A.c0
+    api = OpenAPI("/", with_schema_discriminated_union_extends)