CHanges:

- add hook and fix registering to new registry

edgy/contrib/autoreflection/metaclasses.py

@@ -69,24 +69,14 @@ def __new__(
         name: str,
         bases: tuple[type, ...],
         attrs: dict[str, Any],
-        skip_registry: bool = False,
         meta_info_class: type[AutoReflectionMetaInfo] = AutoReflectionMetaInfo,
         **kwargs: Any,
     ) -> Any:
-        new_model = super().__new__(
+        return super().__new__(
             cls,
             name,
             bases,
             attrs,
             meta_info_class=meta_info_class,
-            skip_registry=True,
             **kwargs,
         )
-        if (
-            not skip_registry
-            and isinstance(new_model.meta, AutoReflectionMetaInfo)
-            and not new_model.meta.abstract
-            and new_model.meta.registry
-        ):
-            new_model.meta.registry.pattern_models[new_model.__name__] = new_model
-        return new_model

edgy/contrib/autoreflection/models.py

@@ -1,9 +1,18 @@
-from typing import ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar
 
 import edgy
 
 from .metaclasses import AutoReflectionMeta, AutoReflectionMetaInfo
 
+if TYPE_CHECKING:
+    from edgy.core.db.models.types import BaseModelType
+
 
 class AutoReflectModel(edgy.ReflectModel, metaclass=AutoReflectionMeta):
     meta: ClassVar[AutoReflectionMetaInfo]
+
+    @classmethod
+    def real_add_to_registry(cls: type["BaseModelType"], **kwargs: Any) -> type["BaseModelType"]:
+        if isinstance(cls.meta, AutoReflectionMetaInfo):
+            kwargs.setdefault("registry_type_name", "pattern_models")
+        return super().real_add_to_registry(**kwargs)

edgy/contrib/multi_tenancy/base.py

@@ -1,8 +1,11 @@
-from typing import ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from edgy.contrib.multi_tenancy.metaclasses import BaseTenantMeta, TenantMeta
 from edgy.core.db.models.model import Model
 
+if TYPE_CHECKING:
+    from edgy.core.db.models.types import BaseModelType
+
 
 class TenantModel(Model, metaclass=BaseTenantMeta):
     """
@@ -16,3 +19,24 @@ class TenantModel(Model, metaclass=BaseTenantMeta):
     """
 
     meta: ClassVar[TenantMeta] = TenantMeta(None, abstract=True)
+
+    @classmethod
+    def real_add_to_registry(cls: type["BaseModelType"], **kwargs: Any) -> type["BaseModelType"]:
+        result = super().real_add_to_registry(**kwargs)
+
+        if (
+            cls.meta.registry
+            and cls.meta.is_tenant
+            and not cls.meta.abstract
+            and not cls.__is_proxy_model__
+        ):
+            assert (
+                cls.__reflected__ is False
+            ), "Reflected models are not compatible with multi_tenancy"
+
+            if not cls.meta.register_default:
+                # remove from models
+                cls.meta.registry.models.pop(cls.__name__, None)
+            cls.meta.registry.tenant_models[cls.__name__] = cls
+
+        return result

edgy/contrib/multi_tenancy/metaclasses.py

@@ -1,10 +1,13 @@
-from typing import Any, Optional, cast
+from typing import TYPE_CHECKING, Any, Literal, Optional, Union, cast
 
 from edgy.core.db.models.metaclasses import (
     BaseModelMeta,
     MetaInfo,
 )
 
+if TYPE_CHECKING:
+    from edgy.core.connection.database import Database
+
 
 def _check_model_inherited_tenancy(bases: tuple[type, ...]) -> bool:
     for base in bases:
@@ -42,23 +45,27 @@ class BaseTenantMeta(BaseModelMeta):
     your own tenant model using the `is_tenant` inside the `Meta` object.
     """
 
-    def __new__(cls, name: str, bases: tuple[type, ...], attrs: Any, **kwargs: Any) -> Any:
-        new_model = super().__new__(cls, name, bases, attrs, meta_info_class=TenantMeta, **kwargs)
+    def __new__(
+        cls,
+        name: str,
+        bases: tuple[type, ...],
+        attrs: Any,
+        on_conflict: Literal["error", "replace", "keep"] = "error",
+        **kwargs: Any,
+    ) -> Any:
+        database: Union[Literal["keep"], None, Database, bool] = attrs.get("database", "keep")
+        new_model = super().__new__(
+            cls, name, bases, attrs, meta_info_class=TenantMeta, skip_registry=True, **kwargs
+        )
         if new_model.meta.is_tenant is None:
             new_model.meta.is_tenant = _check_model_inherited_tenancy(bases)
 
         if (
             new_model.meta.registry
-            and new_model.meta.is_tenant
             and not new_model.meta.abstract
             and not new_model.__is_proxy_model__
         ):
-            assert (
-                new_model.__reflected__ is False
-            ), "Reflected models are not compatible with multi_tenancy"
-
-            if not new_model.meta.register_default:
-                # remove from models
-                new_model.meta.registry.models.pop(new_model.__name__, None)
-            new_model.meta.registry.tenant_models[new_model.__name__] = new_model
+            new_model.add_to_registry(
+                new_model.meta.registry, on_conflict=on_conflict, database=database
+            )
         return new_model

edgy/core/connection/registry.py

@@ -180,9 +180,7 @@ def __copy__(self) -> "Registry":
                 (
                     (
                         key,
-                        val.copy_edgy_model(
-                            registry=_copy if registry_type in {"models", "reflected"} else None
-                        ),
+                        val.copy_edgy_model(registry=_copy),
                     )
                     for key, val in getattr(self, registry_type).items()
                     if not val.meta.no_copy and key not in dict_models
@@ -531,7 +529,9 @@ async def _connect_and_init(self, name: Union[str, None], database: "Database")
                     new_name = pattern_model.meta.template(table)
                     old_model: Optional[type[BaseModelType]] = None
                     with contextlib.suppress(LookupError):
-                        old_model = self.get_model(new_name)
+                        old_model = self.get_model(
+                            new_name, include_content_type_attr=False, exclude=("pattern_models",)
+                        )
                     if old_model is not None:
                         raise Exception(
                             f"Conflicting model: {old_model.__name__} with pattern model: {pattern_model.__name__}"

edgy/core/db/models/mixins/db.py

@@ -168,12 +168,13 @@ class DatabaseMixin:
     _removed_copy_keys: ClassVar[set[str]] = _removed_copy_keys
 
     @classmethod
-    def add_to_registry(
+    def real_add_to_registry(
         cls: type[BaseModelType],
+        *,
         registry: Registry,
+        registry_type_name: str = "models",
         name: str = "",
         database: Union[bool, Database, Literal["keep"]] = "keep",
-        *,
         replace_related_field: Union[
             bool,
             type[BaseModelType],
@@ -182,6 +183,7 @@ def add_to_registry(
         ] = False,
         on_conflict: Literal["keep", "replace", "error"] = "error",
     ) -> type[BaseModelType]:
+        """For customizations."""
         # when called if registry is not set
         cls.meta.registry = registry
         if database is True:
@@ -216,33 +218,57 @@ def add_to_registry(
                                 'setting "on_conflict" to either "keep" or "replace".'
                             )
                         )
-            if getattr(cls, "__reflected__", False):
-                registry.reflected[cls.__name__] = cls
-            else:
-                registry.models[cls.__name__] = cls
-            # after registrating the own model
-            for value in list(meta.fields.values()):
-                if isinstance(value, BaseManyToManyForeignKeyField):
-                    m2m_registry: Registry = value.target_registry
-                    with contextlib.suppress(Exception):
-                        m2m_registry = cast("Registry", value.target.registry)
-
-                    def create_through_model(x: Any, field: BaseFieldType = value) -> None:
-                        # we capture with field = ... the variable
-                        field.create_through_model(replace_related_field=replace_related_field)
-
-                    m2m_registry.register_callback(value.to, create_through_model, one_time=True)
-            # Sets the foreign key fields
-            if meta.foreign_key_fields:
-                _set_related_name_for_foreign_keys(
-                    meta, cls, replace_related_field=replace_related_field
-                )
-            registry.execute_model_callbacks(cls)
+            if registry_type_name:
+                registry_dict = getattr(registry, registry_type_name)
+                registry_dict[cls.__name__] = cls
+                # after registrating the own model
+                for value in list(meta.fields.values()):
+                    if isinstance(value, BaseManyToManyForeignKeyField):
+                        m2m_registry: Registry = value.target_registry
+                        with contextlib.suppress(Exception):
+                            m2m_registry = cast("Registry", value.target.registry)
+
+                        def create_through_model(x: Any, field: BaseFieldType = value) -> None:
+                            # we capture with field = ... the variable
+                            field.create_through_model(replace_related_field=replace_related_field)
+
+                        m2m_registry.register_callback(
+                            value.to, create_through_model, one_time=True
+                        )
+                # Sets the foreign key fields
+                if meta.foreign_key_fields:
+                    _set_related_name_for_foreign_keys(
+                        meta, cls, replace_related_field=replace_related_field
+                    )
+                registry.execute_model_callbacks(cls)
 
         # finalize
         cls.model_rebuild(force=True)
         return cls
 
+    @classmethod
+    def add_to_registry(
+        cls: type[BaseModelType],
+        registry: Registry,
+        name: str = "",
+        database: Union[bool, Database, Literal["keep"]] = "keep",
+        *,
+        replace_related_field: Union[
+            bool,
+            type[BaseModelType],
+            tuple[type[BaseModelType], ...],
+            list[type[BaseModelType]],
+        ] = False,
+        on_conflict: Literal["keep", "replace", "error"] = "error",
+    ) -> type[BaseModelType]:
+        return cls.real_add_to_registry(
+            registry=registry,
+            name=name,
+            database=database,
+            replace_related_field=replace_related_field,
+            on_conflict=on_conflict,
+        )
+
     def get_active_instance_schema(
         self, check_schema: bool = True, check_tenant: bool = True
     ) -> Union[str, None]:
@@ -295,6 +321,7 @@ def copy_edgy_model(
             __metadata__=meta_info,
             __bases__=cls.__bases__,
             skip_registry=True,
+            **kwargs,
         )
         # should also allow masking database with None
         if hasattr(cls, "database"):

edgy/core/db/models/mixins/reflection.py

@@ -9,6 +9,7 @@
 if TYPE_CHECKING:
     from edgy import Registry
     from edgy.core.connection.database import Database
+    from edgy.core.db.models.types import BaseModelType
 
 
 class ReflectedModelMixin:
@@ -19,6 +20,11 @@ class ReflectedModelMixin:
 
     __reflected__: ClassVar[bool] = True
 
+    @classmethod
+    def real_add_to_registry(cls: type["BaseModelType"], **kwargs: Any) -> type["BaseModelType"]:
+        kwargs.setdefault("registry_type_name", "reflected")
+        return cast(type["BaseModelType"], super().real_add_to_registry(**kwargs))
+
     @classmethod
     def build(
         cls, schema: Optional[str] = None, metadata: Optional[sqlalchemy.MetaData] = None