🐛 Do not register multiple times same child schemas (#188)

perdy · perdy · commit ae72a1873e50 · 2025-10-14T23:55:14.000+02:00
diff --git a/flama/schemas/generator.py b/flama/schemas/generator.py
@@ -7,7 +7,7 @@
 
 import yaml
 
-from flama import routing, schemas, types, url
+from flama import exceptions, routing, schemas, types, url
 from flama.schemas import Schema, openapi
 from flama.schemas.data_structures import Parameter
 
@@ -205,20 +205,20 @@ def register(self, schema: schemas.Schema, name: t.Optional[str] = None) -> int:
         :return: Schema ID.
         """
         if schema in self:
-            raise ValueError("Schema is already registered.")
+            raise exceptions.ApplicationError(f"Schema '{schema}' is already registered.")
 
         s = schemas.Schema(schema)
 
         try:
             schema_name = name or s.name
         except ValueError as e:  # pragma: no cover
-            raise ValueError("Cannot infer schema name.") from e
+            raise exceptions.ApplicationError("Cannot infer schema name.") from e
 
         schema_instance = s.unique_schema
         schema_id = id(schema_instance)
         self[schema_id] = SchemaInfo(name=schema_name, schema=schema_instance)
 
-        for child_schema in [schemas.Schema(x) for x in s.nested_schemas() if x not in self]:
+        for child_schema in (schemas.Schema(x) for x in s.nested_schemas() if x not in self):
             self.register(schema=child_schema.schema, name=child_schema.name)
 
         return schema_id
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -165,9 +165,6 @@ def model_cls(self, framework: str):
         return self._models_cls[framework]
 
     def _sklearn(self):
-        assert np
-        assert sklearn
-
         model = sklearn.neural_network.MLPClassifier(activation="tanh", max_iter=2000, hidden_layer_sizes=(10,))
         model.fit(
             np.array([[0, 0], [0, 1], [1, 0], [1, 1]]),
@@ -176,9 +173,6 @@ def _sklearn(self):
         return model, sklearn.neural_network.MLPClassifier
 
     def _sklearn_pipeline(self):
-        assert np
-        assert sklearn
-
         model = sklearn.neural_network.MLPClassifier(activation="tanh", max_iter=2000, hidden_layer_sizes=(10,))
         numerical_transformer = sklearn.pipeline.Pipeline(
             [
@@ -205,9 +199,6 @@ def _sklearn_pipeline(self):
         return pipeline, sklearn.pipeline.Pipeline
 
     def _tensorflow(self):
-        assert np
-        assert tf
-
         model = tf.keras.models.Sequential(
             [
                 tf.keras.Input((2,)),
@@ -227,9 +218,6 @@ def _tensorflow(self):
         return model, tf.keras.models.Sequential
 
     def _torch(self):
-        assert np
-        assert torch
-
         class Model(torch.nn.Module):
             def __init__(self):
                 super().__init__()
diff --git a/tests/schemas/conftest.py b/tests/schemas/conftest.py
@@ -74,6 +74,45 @@ def bar_optional_schema(app, foo_schema):
     return namedtuple("BarOptionalSchema", ("schema", "name"))(schema=schema, name=name)
 
 
+@pytest.fixture(scope="function")
+def bar_multiple_schema(app, foo_schema):
+    child_schema = foo_schema.schema
+    if app.schema.schema_library.name == "pydantic":
+        schema = pydantic.create_model(
+            "BarMultiple",
+            first=(t.Union[child_schema, None], None),
+            second=(t.Union[child_schema, None], None),
+            __module__="pydantic.main",
+        )
+        name = "pydantic.main.BarOptional"
+    elif app.schema.schema_library.name == "typesystem":
+        schema = typesystem.Schema(
+            title="BarOptional",
+            fields={
+                "first": typesystem.Reference(
+                    to="Foo", definitions=typesystem.Definitions({"Foo": child_schema}), allow_null=True, default=None
+                ),
+                "second": typesystem.Reference(
+                    to="Foo", definitions=typesystem.Definitions({"Foo": child_schema}), allow_null=True, default=None
+                ),
+            },
+        )
+        name = "typesystem.schemas.BarOptional"
+    elif app.schema.schema_library.name == "marshmallow":
+        schema = type(
+            "BarOptional",
+            (marshmallow.Schema,),
+            {
+                "first": marshmallow.fields.Nested(child_schema(), required=False, dump_default=None, allow_none=True),
+                "second": marshmallow.fields.Nested(child_schema(), required=False, dump_default=None, allow_none=True),
+            },
+        )
+        name = "abc.BarOptional"
+    else:
+        raise ValueError(f"Wrong schema lib: {app.schema.schema_library.name}")
+    return namedtuple("BarMultipleSchema", ("schema", "name"))(schema=schema, name=name)
+
+
 @pytest.fixture(scope="function")
 def bar_list_schema(app, foo_schema):
     child_schema = foo_schema.schema
@@ -131,5 +170,12 @@ def bar_dict_schema(app, foo_schema):
 
 
 @pytest.fixture(scope="function")
-def schemas(foo_schema, bar_schema, bar_list_schema, bar_dict_schema):
-    return {"Foo": foo_schema, "Bar": bar_schema, "BarList": bar_list_schema, "BarDict": bar_dict_schema}
+def schemas(foo_schema, bar_schema, bar_optional_schema, bar_multiple_schema, bar_list_schema, bar_dict_schema):
+    return {
+        "Foo": foo_schema,
+        "Bar": bar_schema,
+        "BarOptional": bar_optional_schema,
+        "BarMultiple": bar_multiple_schema,
+        "BarList": bar_list_schema,
+        "BarDict": bar_dict_schema,
+    }
diff --git a/tests/schemas/test_generator.py b/tests/schemas/test_generator.py
@@ -8,7 +8,7 @@
 import typesystem
 import typesystem.fields
 
-from flama import Flama, schemas
+from flama import Flama, exceptions, schemas
 from flama.endpoints import HTTPEndpoint
 from flama.schemas import openapi
 from flama.schemas.generator import SchemaRegistry
@@ -782,30 +782,57 @@ def test_used(self, registry, schemas, spec, operation, register_schemas, output
         assert set(registry.used(spec).keys()) == expected_output
 
     @pytest.mark.parametrize(
-        ["schema", "explicit_name", "output"],
+        ["cases"],
         [
-            pytest.param("Foo", "Foo", {"Foo": "Foo"}, id="explicit_name"),
-            pytest.param("Foo", None, {"Foo": None}, id="infer_name"),
-            pytest.param("Bar", "Bar", {"Bar": "Bar"}, id="nested_schemas"),
+            pytest.param(
+                [
+                    ("Foo", "Foo", {"Foo": "Foo"}, None),
+                ],
+                id="explicit_name",
+            ),
+            pytest.param(
+                [
+                    ("Foo", None, {"Foo": None}, None),
+                ],
+                id="infer_name",
+            ),
+            pytest.param(
+                [
+                    ("Bar", "Bar", {"Bar": "Bar"}, None),
+                ],
+                id="nested_schemas",
+            ),
+            pytest.param(
+                [
+                    ("Foo", "Foo", {"Foo": "Foo"}, None),
+                    ("Foo", None, None, (exceptions.ApplicationError, r"Schema '.*' is already registered.")),
+                ],
+                id="error_already_registered",
+            ),
+            pytest.param(
+                [
+                    ("BarMultiple", "BarMultiple", {}, None),
+                ],
+                id="multiple_child_schema",
+            ),
         ],
     )
-    def test_register(self, registry, schemas, schema, explicit_name, output):
-        schema, name = schemas[schema]
-        expected_name = name if not explicit_name else explicit_name
-        exception = (
-            contextlib.ExitStack() if expected_name else pytest.raises(ValueError, match="Cannot infer schema name.")
-        )
-        with exception:
-            registry.register(schema, name=explicit_name)
-            for s, n in output.items():
-                assert schemas[s].schema in registry
-                assert registry[schemas[s].schema].name == (n or schemas[s].name)
+    def test_register(self, registry, schemas, cases):
+        for schema_key, explicit_name, output, exception in cases:
+            schema, name = schemas[schema_key]
 
-    def test_register_already_registered(self, registry, foo_schema):
-        schema = foo_schema.schema
-        registry.register(schema, name="Foo")
-        with pytest.raises(ValueError, match="Schema is already registered."):
-            registry.register(schema, name="Foo")
+            if explicit_name is None and name is None:
+                exception = pytest.raises(ValueError, match="Cannot infer schema name.")
+            elif exception is not None:
+                exception = pytest.raises(exception[0], match=exception[1])
+            else:
+                exception = contextlib.ExitStack()
+
+            with exception:
+                registry.register(schema, name=explicit_name)
+                for s, n in output.items():
+                    assert schemas[s].schema in registry
+                    assert registry[schemas[s].schema].name == (n or schemas[s].name)
 
     @pytest.mark.parametrize(
         ["multiple", "result"],
