Fix migration bug 2 & copying (#227)

Changes:

- fix lazyness (split in two keys)
- make model_rebuild in fields lazy
- add content type shortcut for get_model
- improved content type tests
- updated migrate test
- fix copying

docs/models.md

@@ -164,7 +164,7 @@ to copy a model class and optionally add it to an other registry.
 
 You can add it to a registry later by using:
 
-`model_class.add_to_registry(registry, name="")`
+`model_class.add_to_registry(registry, name="", database=None, replace_related_field=False)`
 
 In fact the last method is called when the registry parameter of `copy_edgy_model` is not `None`.
 

docs/release-notes.md

@@ -14,16 +14,23 @@ hide:
 - Global constraints via meta.
 - Allow functional indexes.
 - Expose further parameters for UniqueConstraints.
+- `no_copy` attribute for fields.
 
 ### Changes
 
 - Breaking: Factories pass now the kwargs as dict to get_pydantic_type, get_column_type, get_constraints.
   This allows now modifying the arguments passed down to the field.
+- Breaking: init_fields_mapping doesn't initializes the field stats anymore.
+- Breaking: model rebuilds are executed lazily when calling init_fields_mapping not when assigning fields manually anymore.
 
 ### Fixed
 
 - Indexes and unique_together worked only for fields with columns of the same name.
 - MigrateConfig has no get_registry_copy.
+- Migrations have duplicate fks and crash.
+- ContentTypes were not copyable.
+- VirtualCascade was not automatically enabled.
+- Improve lazyness by splitting in two variable sets.
 
 ## 0.21.2
 

edgy/contrib/contenttypes/fields.py

@@ -63,10 +63,8 @@ def __new__(  # type: ignore
         to: Union[type["BaseModelType"], str] = "ContentType",
         on_delete: str = CASCADE,
         no_constraint: bool = False,
-        delete_orphan: bool = True,
-        default: Any = lambda owner: owner.meta.registry.get_model("ContentType")(
-            name=owner.__name__
-        ),
+        remove_referenced: bool = True,
+        default: Any = lambda owner: owner.meta.registry.content_type(name=owner.__name__),
         **kwargs: Any,
     ) -> "BaseFieldType":
         return super().__new__(
@@ -75,7 +73,7 @@ def __new__(  # type: ignore
             default=default,
             on_delete=on_delete,
             no_constraint=no_constraint,
-            delete_orphan=delete_orphan,
+            remove_referenced=remove_referenced,
             **kwargs,
         )
 

edgy/contrib/contenttypes/models.py

@@ -19,7 +19,7 @@ class Meta:
     name: str = edgy.fields.CharField(max_length=100, default="", index=True)
     # set also the schema for tenancy support
     schema_name: str = edgy.CharField(max_length=63, null=True, index=True)
-    # can be a hash or similar. For checking collisions cross domain
+    # can be a hash or similar. Usefull for checking collisions cross domain
     collision_key: str = edgy.fields.CharField(max_length=255, null=True, unique=True)
 
     async def get_instance(self) -> edgy.Model:
@@ -34,7 +34,10 @@ async def delete(
         self, skip_post_delete_hooks: bool = False, remove_referenced_call: bool = False
     ) -> None:
         reverse_name = f"reverse_{self.name.lower()}"
-        query = cast("QuerySet", getattr(self, reverse_name))
+        referenced_obs = cast("QuerySet", getattr(self, reverse_name))
         await super().delete(skip_post_delete_hooks=skip_post_delete_hooks)
-        if not remove_referenced_call and self.no_constraint:
-            await query.using(schema=self.schema_name).delete()
+        if (
+            not remove_referenced_call
+            and self.meta.fields[reverse_name].foreign_key.force_cascade_deletion_relation
+        ):
+            await referenced_obs.using(schema=self.schema_name).delete()

edgy/core/connection/registry.py

@@ -3,7 +3,7 @@
 import re
 import warnings
 from collections import defaultdict
-from collections.abc import Mapping, Sequence
+from collections.abc import Sequence
 from copy import copy as shallow_copy
 from functools import cached_property, partial
 from types import TracebackType
@@ -132,7 +132,7 @@ def __init__(
             defaultdict(list)
         )
 
-        self.extra: Mapping[str, Database] = {
+        self.extra: dict[str, Database] = {
             k: v if isinstance(v, Database) else Database(v) for k, v in extra.items()
         }
         self.metadata_by_url = MetaDataByUrlDict(registry=self)
@@ -141,28 +141,37 @@ def __init__(
             self._set_content_type(with_content_type)
 
     def __copy__(self) -> "Registry":
-        _copy = Registry(self.database)
-        _copy.extra = self.extra
-        _copy.models = {key: val.copy_edgy_model(_copy) for key, val in self.models.items()}
-        _copy.reflected = {key: val.copy_edgy_model(_copy) for key, val in self.reflected.items()}
-        _copy.tenant_models = {
-            key: val.copy_edgy_model(_copy) for key, val in self.tenant_models.items()
-        }
-        _copy.pattern_models = {
-            key: val.copy_edgy_model(_copy) for key, val in self.pattern_models.items()
-        }
-        _copy.dbs_reflected = set(self.dbs_reflected)
+        content_type: Union[bool, type[BaseModelType]] = False
         if self.content_type is not None:
             try:
-                _copy.content_type = self.get_model("ContentType")
+                content_type2 = content_type = self.get_model(
+                    "ContentType", include_content_type_attr=False
+                ).copy_edgy_model()
+                # cleanup content_type copy
+                for field_name in list(content_type2.meta.fields.keys()):
+                    if field_name.startswith("reverse_"):
+                        del content_type2.meta.fields[field_name]
             except LookupError:
-                _copy.content_type = self.content_type
-            # init callbacks
-            _copy._set_content_type(_copy.content_type)
+                content_type = self.content_type
+        _copy = Registry(
+            self.database, with_content_type=content_type, schema=self.db_schema, extra=self.extra
+        )
+        for i in ["models", "reflected", "tenant_models", "pattern_models"]:
+            dict_models = getattr(_copy, i)
+            dict_models.update(
+                (
+                    (key, val.copy_edgy_model(_copy))
+                    for key, val in getattr(self, i).items()
+                    if key not in dict_models
+                )
+            )
+        _copy.dbs_reflected = set(self.dbs_reflected)
         return _copy
 
     def _set_content_type(
-        self, with_content_type: Union[Literal[True], type["BaseModelType"]]
+        self,
+        with_content_type: Union[Literal[True], type["BaseModelType"]],
+        old_content_type_to_replace: Optional[type["BaseModelType"]] = None,
     ) -> None:
         from edgy.contrib.contenttypes.fields import BaseContentTypeFieldField, ContentTypeField
         from edgy.contrib.contenttypes.models import ContentType
@@ -197,31 +206,52 @@ def callback(model_class: type["BaseModelType"]) -> None:
             # they are not updated, despite this shouldn't happen anyway
             if issubclass(model_class, ContentType):
                 return
-            # skip if is explicit set
+            # skip if is explicit set or remove when copying
             for field in model_class.meta.fields.values():
                 if isinstance(field, BaseContentTypeFieldField):
+                    if (
+                        old_content_type_to_replace is not None
+                        and field.target is old_content_type_to_replace
+                    ):
+                        field.target_registry = self
+                        field.target = real_content_type
+                        # simply overwrite
+                        real_content_type.meta.fields[field.related_name] = RelatedField(
+                            name=field.related_name,
+                            foreign_key_name=field.name,
+                            related_from=model_class,
+                            owner=real_content_type,
+                        )
                     return
+
             # e.g. exclude field
-            if "content_type" not in model_class.meta.fields:
-                related_name = f"reverse_{model_class.__name__.lower()}"
-                assert (
-                    related_name not in real_content_type.meta.fields
-                ), f"duplicate model name: {model_class.__name__}"
-                model_class.meta.fields["content_type"] = cast(
-                    "BaseFieldType",
-                    ContentTypeField(
-                        name="content_type",
-                        owner=model_class,
-                        to=real_content_type,
-                        no_constraint=real_content_type.no_constraint,
-                    ),
-                )
-                real_content_type.meta.fields[related_name] = RelatedField(
-                    name=related_name,
-                    foreign_key_name="content_type",
-                    related_from=model_class,
-                    owner=real_content_type,
-                )
+            if "content_type" in model_class.meta.fields:
+                return
+            related_name = f"reverse_{model_class.__name__.lower()}"
+            assert (
+                related_name not in real_content_type.meta.fields
+            ), f"duplicate model name: {model_class.__name__}"
+
+            field_args: dict[str, Any] = {
+                "name": "content_type",
+                "owner": model_class,
+                "to": real_content_type,
+                "no_constraint": real_content_type.no_constraint,
+                "no_copy": True,
+            }
+            if model_class.meta.registry is not real_content_type.meta.registry:
+                field_args["relation_has_post_delete_callback"] = True
+                field_args["force_cascade_deletion_relation"] = True
+            model_class.meta.fields["content_type"] = cast(
+                "BaseFieldType",
+                ContentTypeField(**field_args),
+            )
+            real_content_type.meta.fields[related_name] = RelatedField(
+                name=related_name,
+                foreign_key_name="content_type",
+                related_from=model_class,
+                owner=real_content_type,
+            )
 
         self.register_callback(None, callback, one_time=False)
 
@@ -248,7 +278,15 @@ def metadata(self) -> sqlalchemy.MetaData:
         )
         return self.metadata_by_name[None]
 
-    def get_model(self, model_name: str) -> type["BaseModelType"]:
+    def get_model(
+        self, model_name: str, *, include_content_type_attr: bool = True
+    ) -> type["BaseModelType"]:
+        if (
+            include_content_type_attr
+            and model_name == "ContentType"
+            and self.content_type is not None
+        ):
+            return self.content_type
         if model_name in self.models:
             return self.models[model_name]
         elif model_name in self.reflected:

edgy/core/db/fields/base.py

@@ -9,6 +9,7 @@
     Literal,
     Optional,
     Union,
+    cast,
 )
 
 import sqlalchemy
@@ -418,11 +419,18 @@ class BaseForeignKey(RelationshipField):
     # only useful if related_name = False because otherwise it gets overwritten
     reverse_name: str = ""
 
-    @cached_property
+    @property
     def target_registry(self) -> "Registry":
         """Registry searched in case to is a string"""
-        assert self.owner.meta.registry, "no registry found neither 'target_registry' set"
-        return self.owner.meta.registry
+
+        if not hasattr(self, "_target_registry"):
+            assert self.owner.meta.registry, "no registry found neither 'target_registry' set"
+            return self.owner.meta.registry
+        return cast("Registry", self._target_registry)
+
+    @target_registry.setter
+    def target_registry(self, value: Any) -> None:
+        self._target_registry = value
 
     @property
     def target(self) -> Any:

edgy/core/db/fields/foreign_keys.py

@@ -12,7 +12,7 @@
 import sqlalchemy
 from pydantic import BaseModel
 
-from edgy.core.db.constants import CASCADE, SET_DEFAULT, SET_NULL
+from edgy.core.db.constants import SET_DEFAULT, SET_NULL
 from edgy.core.db.context_vars import CURRENT_PHASE
 from edgy.core.db.fields.base import BaseForeignKey
 from edgy.core.db.fields.factories import ForeignKeyFieldFactory
@@ -70,10 +70,6 @@ def __init__(
             )
         if self.on_delete == SET_NULL and not self.null:
             terminal.write_warning("Declaring on_delete `SET NULL` but null is False.")
-        if self.force_cascade_deletion_relation or (
-            self.on_delete == CASCADE and self.no_constraint
-        ):
-            self.relation_has_post_delete_callback = True
 
     async def _notset_post_delete_callback(self, value: Any, instance: "BaseModelType") -> None:
         value = self.expand_relationship(value)
@@ -104,9 +100,7 @@ async def pre_save_callback(
     def get_relation(self, **kwargs: Any) -> ManyRelationProtocol:
         if self.relation_fn is not None:
             return self.relation_fn(**kwargs)
-        if self.force_cascade_deletion_relation or (
-            self.on_delete == CASCADE and self.no_constraint
-        ):
+        if self.force_cascade_deletion_relation:
             relation: Any = VirtualCascadeDeletionSingleRelation
         else:
             relation = SingleRelation

edgy/core/db/fields/types.py

@@ -47,6 +47,7 @@ class ColumnDefinitionModel(
 
 
 class BaseFieldDefinitions:
+    no_copy: bool = False
     read_only: bool = False
     inject_default_on_partial_update: bool = False
     inherit: bool = True

edgy/core/db/models/base.py

@@ -163,7 +163,7 @@ def identifying_db_fields(self) -> Any:
         """The columns used for loading, can be set per instance defaults to pknames"""
         return self.pkcolumns
 
-    @cached_property
+    @property
     def proxy_model(self) -> type[Model]:
         return self.__class__.proxy_model  # type: ignore