From 32695ffa1d84e203c315d6a1ae296b1082e555ad Mon Sep 17 00:00:00 2001 From: collerek Date: Fri, 19 Mar 2021 14:22:31 +0100 Subject: [PATCH] switch from class to instance fro fields --- ormar/fields/base.py | 274 +++++++++++--------------- ormar/fields/foreign_key.py | 120 ++++++----- ormar/fields/many_to_many.py | 86 ++++---- ormar/fields/model_fields.py | 23 ++- ormar/fields/sqlalchemy_encrypted.py | 2 +- ormar/fields/through_field.py | 3 +- ormar/models/helpers/models.py | 2 +- ormar/models/helpers/pydantic.py | 57 +----- ormar/models/helpers/relations.py | 22 +-- ormar/models/helpers/sqlalchemy.py | 16 +- ormar/models/helpers/validation.py | 10 +- ormar/models/metaclass.py | 28 ++- ormar/models/mixins/alias_mixin.py | 4 +- ormar/models/mixins/prefetch_mixin.py | 4 +- ormar/models/mixins/relation_mixin.py | 9 +- ormar/models/model_row.py | 10 +- ormar/models/newbasemodel.py | 8 +- ormar/queryset/prefetch_query.py | 16 +- ormar/queryset/utils.py | 4 +- ormar/relations/alias_manager.py | 4 +- ormar/relations/relation_manager.py | 12 +- ormar/relations/utils.py | 4 +- tests/test_inheritance_concrete.py | 4 +- tests/test_inheritance_mixins.py | 12 +- tests/test_model_definition.py | 2 +- tests/test_models.py | 4 +- 26 files changed, 329 insertions(+), 411 deletions(-) diff --git a/ormar/fields/base.py b/ormar/fields/base.py index cecb726..f96a064 100644 --- a/ormar/fields/base.py +++ b/ormar/fields/base.py @@ -1,7 +1,7 @@ -from typing import Any, List, Optional, TYPE_CHECKING, Type, Union +from typing import Any, Dict, List, Optional, TYPE_CHECKING, Type, Union import sqlalchemy -from pydantic import Field, Json, typing +from pydantic import Json, typing from pydantic.fields import FieldInfo, Required, Undefined import ormar # noqa I101 @@ -28,44 +28,62 @@ class BaseField(FieldInfo): to pydantic field types like ConstrainedStr """ - __type__ = None - related_name = None + def __init__(self, **kwargs: Any) -> None: + self.__type__: type = kwargs.pop("__type__", None) + self.related_name = kwargs.pop("related_name", None) - column_type: sqlalchemy.Column - constraints: List = [] - name: str - alias: str + self.column_type: sqlalchemy.Column = kwargs.pop("column_type", None) + self.constraints: List = kwargs.pop("constraints", list()) + self.name: str = kwargs.pop("name", None) + self.db_alias: str = kwargs.pop("alias", None) - primary_key: bool - autoincrement: bool - nullable: bool - index: bool - unique: bool - pydantic_only: bool - choices: typing.Sequence + self.primary_key: bool = kwargs.pop("primary_key", False) + self.autoincrement: bool = kwargs.pop("autoincrement", False) + self.nullable: bool = kwargs.pop("nullable", False) + self.index: bool = kwargs.pop("index", False) + self.unique: bool = kwargs.pop("unique", False) + self.pydantic_only: bool = kwargs.pop("pydantic_only", False) + self.choices: typing.Sequence = kwargs.pop("choices", False) - virtual: bool = False # ManyToManyFields and reverse ForeignKeyFields - is_multi: bool = False # ManyToManyField - is_relation: bool = False # ForeignKeyField + subclasses - is_through: bool = False # ThroughFields + self.virtual: bool = kwargs.pop( + "virtual", None + ) # ManyToManyFields and reverse ForeignKeyFields + self.is_multi: bool = kwargs.pop("is_multi", None) # ManyToManyField + self.is_relation: bool = kwargs.pop( + "is_relation", None + ) # ForeignKeyField + subclasses + self.is_through: bool = kwargs.pop("is_through", False) # ThroughFields - owner: Type["Model"] - to: Type["Model"] - through: Type["Model"] - self_reference: bool = False - self_reference_primary: Optional[str] = None - orders_by: Optional[List[str]] = None - related_orders_by: Optional[List[str]] = None + self.owner: Type["Model"] = kwargs.pop("owner", None) + self.to: Type["Model"] = kwargs.pop("to", None) + self.through: Type["Model"] = kwargs.pop("through", None) + self.self_reference: bool = kwargs.pop("self_reference", False) + self.self_reference_primary: Optional[str] = kwargs.pop( + "self_reference_primary", None + ) + self.orders_by: Optional[List[str]] = kwargs.pop("orders_by", None) + self.related_orders_by: Optional[List[str]] = kwargs.pop( + "related_orders_by", None + ) - encrypt_secret: str - encrypt_backend: EncryptBackends = EncryptBackends.NONE - encrypt_custom_backend: Optional[Type[EncryptBackend]] = None + self.encrypt_secret: str = kwargs.pop("encrypt_secret", None) + self.encrypt_backend: EncryptBackends = kwargs.pop( + "encrypt_backend", EncryptBackends.NONE + ) + self.encrypt_custom_backend: Optional[Type[EncryptBackend]] = kwargs.pop( + "encrypt_custom_backend", None + ) - default: Any - server_default: Any + self.ormar_default: Any = kwargs.pop("default", None) + self.server_default: Any = kwargs.pop("server_default", None) - @classmethod - def is_valid_uni_relation(cls) -> bool: + for name, value in kwargs.items(): + setattr(self, name, value) + + kwargs.update(self.get_pydantic_default()) + super().__init__(**kwargs) + + def is_valid_uni_relation(self) -> bool: """ Checks if field is a relation definition but only for ForeignKey relation, so excludes ManyToMany fields, as well as virtual ForeignKey @@ -78,10 +96,9 @@ class BaseField(FieldInfo): :return: result of the check :rtype: bool """ - return not cls.is_multi and not cls.virtual + return not self.is_multi and not self.virtual - @classmethod - def get_alias(cls) -> str: + def get_alias(self) -> str: """ Used to translate Model column names to database column names during db queries. @@ -89,73 +106,25 @@ class BaseField(FieldInfo): otherwise field name in ormar/pydantic :rtype: str """ - return cls.alias if cls.alias else cls.name + return self.db_alias if self.db_alias else self.name - @classmethod - def is_valid_field_info_field(cls, field_name: str) -> bool: - """ - Checks if field belongs to pydantic FieldInfo - - used during setting default pydantic values. - Excludes defaults and alias as they are populated separately - (defaults) or not at all (alias) - - :param field_name: field name of BaseFIeld - :type field_name: str - :return: True if field is present on pydantic.FieldInfo - :rtype: bool - """ - return ( - field_name not in ["default", "default_factory", "alias", "allow_mutation"] - and not field_name.startswith("__") - and hasattr(cls, field_name) - and not callable(getattr(cls, field_name)) - ) - - @classmethod - def get_base_pydantic_field_info(cls, allow_null: bool) -> FieldInfo: + def get_pydantic_default(self) -> Dict: """ Generates base pydantic.FieldInfo with only default and optionally required to fix pydantic Json field being set to required=False. Used in an ormar Model Metaclass. - :param allow_null: flag if the default value can be None - or if it should be populated by pydantic Undefined - :type allow_null: bool :return: instance of base pydantic.FieldInfo :rtype: pydantic.FieldInfo """ - base = cls.default_value() + base = self.default_value() if base is None: - base = ( - FieldInfo(default=None) - if (cls.nullable or allow_null) - else FieldInfo(default=Undefined) - ) - if cls.__type__ == Json and base.default is Undefined: - base.default = Required + base = dict(default=None) if self.nullable else dict(default=Undefined) + if self.__type__ == Json and base.get("default") is Undefined: + base["default"] = Required return base - @classmethod - def convert_to_pydantic_field_info(cls, allow_null: bool = False) -> FieldInfo: - """ - Converts a BaseField into pydantic.FieldInfo - that is later easily processed by pydantic. - Used in an ormar Model Metaclass. - - :param allow_null: flag if the default value can be None - or if it should be populated by pydantic Undefined - :type allow_null: bool - :return: actual instance of pydantic.FieldInfo with all needed fields populated - :rtype: pydantic.FieldInfo - """ - base = cls.get_base_pydantic_field_info(allow_null=allow_null) - for attr_name in FieldInfo.__dict__.keys(): - if cls.is_valid_field_info_field(attr_name): - setattr(base, attr_name, cls.__dict__.get(attr_name)) - return base - - @classmethod - def default_value(cls, use_server: bool = False) -> Optional[FieldInfo]: + def default_value(self, use_server: bool = False) -> Optional[Dict]: """ Returns a FieldInfo instance with populated default (static) or default_factory (function). @@ -173,17 +142,20 @@ class BaseField(FieldInfo): which is returning a FieldInfo instance :rtype: Optional[pydantic.FieldInfo] """ - if cls.is_auto_primary_key(): - return Field(default=None) - if cls.has_default(use_server=use_server): - default = cls.default if cls.default is not None else cls.server_default + if self.is_auto_primary_key(): + return dict(default=None) + if self.has_default(use_server=use_server): + default = ( + self.ormar_default + if self.ormar_default is not None + else self.server_default + ) if callable(default): - return Field(default_factory=default) - return Field(default=default) + return dict(default_factory=default) + return dict(default=default) return None - @classmethod - def get_default(cls, use_server: bool = False) -> Any: # noqa CCR001 + def get_default(self, use_server: bool = False) -> Any: # noqa CCR001 """ Return default value for a field. If the field is Callable the function is called and actual result is returned. @@ -195,18 +167,17 @@ class BaseField(FieldInfo): :return: default value for the field if set, otherwise implicit None :rtype: Any """ - if cls.has_default(): + if self.has_default(): default = ( - cls.default - if cls.default is not None - else (cls.server_default if use_server else None) + self.ormar_default + if self.ormar_default is not None + else (self.server_default if use_server else None) ) if callable(default): default = default() return default - @classmethod - def has_default(cls, use_server: bool = True) -> bool: + def has_default(self, use_server: bool = True) -> bool: """ Checks if the field has default value set. @@ -216,12 +187,11 @@ class BaseField(FieldInfo): :return: result of the check if default value is set :rtype: bool """ - return cls.default is not None or ( - cls.server_default is not None and use_server + return self.ormar_default is not None or ( + self.server_default is not None and use_server ) - @classmethod - def is_auto_primary_key(cls) -> bool: + def is_auto_primary_key(self) -> bool: """ Checks if field is first a primary key and if it, it's than check if it's set to autoincrement. @@ -230,12 +200,11 @@ class BaseField(FieldInfo): :return: result of the check for primary key and autoincrement :rtype: bool """ - if cls.primary_key: - return cls.autoincrement + if self.primary_key: + return self.autoincrement return False - @classmethod - def construct_constraints(cls) -> List: + def construct_constraints(self) -> List: """ Converts list of ormar constraints into sqlalchemy ForeignKeys. Has to be done dynamically as sqlalchemy binds ForeignKey to the table. @@ -249,15 +218,14 @@ class BaseField(FieldInfo): con.reference, ondelete=con.ondelete, onupdate=con.onupdate, - name=f"fk_{cls.owner.Meta.tablename}_{cls.to.Meta.tablename}" - f"_{cls.to.get_column_alias(cls.to.Meta.pkname)}_{cls.name}", + name=f"fk_{self.owner.Meta.tablename}_{self.to.Meta.tablename}" + f"_{self.to.get_column_alias(self.to.Meta.pkname)}_{self.name}", ) - for con in cls.constraints + for con in self.constraints ] return constraints - @classmethod - def get_column(cls, name: str) -> sqlalchemy.Column: + def get_column(self, name: str) -> sqlalchemy.Column: """ Returns definition of sqlalchemy.Column used in creation of sqlalchemy.Table. Populates name, column type constraints, as well as a number of parameters like @@ -268,24 +236,23 @@ class BaseField(FieldInfo): :return: actual definition of the database column as sqlalchemy requires. :rtype: sqlalchemy.Column """ - if cls.encrypt_backend == EncryptBackends.NONE: + if self.encrypt_backend == EncryptBackends.NONE: column = sqlalchemy.Column( - cls.alias or name, - cls.column_type, - *cls.construct_constraints(), - primary_key=cls.primary_key, - nullable=cls.nullable and not cls.primary_key, - index=cls.index, - unique=cls.unique, - default=cls.default, - server_default=cls.server_default, + self.db_alias or name, + self.column_type, + *self.construct_constraints(), + primary_key=self.primary_key, + nullable=self.nullable and not self.primary_key, + index=self.index, + unique=self.unique, + default=self.ormar_default, + server_default=self.server_default, ) else: - column = cls._get_encrypted_column(name=name) + column = self._get_encrypted_column(name=name) return column - @classmethod - def _get_encrypted_column(cls, name: str) -> sqlalchemy.Column: + def _get_encrypted_column(self, name: str) -> sqlalchemy.Column: """ Returns EncryptedString column type instead of actual column. @@ -294,29 +261,28 @@ class BaseField(FieldInfo): :return: newly defined column :rtype: sqlalchemy.Column """ - if cls.primary_key or cls.is_relation: + if self.primary_key or self.is_relation: raise ModelDefinitionError( "Primary key field and relations fields" "cannot be encrypted!" ) column = sqlalchemy.Column( - cls.alias or name, + self.db_alias or name, EncryptedString( - _field_type=cls, - encrypt_secret=cls.encrypt_secret, - encrypt_backend=cls.encrypt_backend, - encrypt_custom_backend=cls.encrypt_custom_backend, + _field_type=self, + encrypt_secret=self.encrypt_secret, + encrypt_backend=self.encrypt_backend, + encrypt_custom_backend=self.encrypt_custom_backend, ), - nullable=cls.nullable, - index=cls.index, - unique=cls.unique, - default=cls.default, - server_default=cls.server_default, + nullable=self.nullable, + index=self.index, + unique=self.unique, + default=self.ormar_default, + server_default=self.server_default, ) return column - @classmethod def expand_relationship( - cls, + self, value: Any, child: Union["Model", "NewBaseModel"], to_register: bool = True, @@ -339,21 +305,19 @@ class BaseField(FieldInfo): """ return value - @classmethod - def set_self_reference_flag(cls) -> None: + def set_self_reference_flag(self) -> None: """ Sets `self_reference` to True if field to and owner are same model. :return: None :rtype: None """ - if cls.owner is not None and ( - cls.owner == cls.to or cls.owner.Meta == cls.to.Meta + if self.owner is not None and ( + self.owner == self.to or self.owner.Meta == self.to.Meta ): - cls.self_reference = True - cls.self_reference_primary = cls.name + self.self_reference = True + self.self_reference_primary = self.name - @classmethod - def has_unresolved_forward_refs(cls) -> bool: + def has_unresolved_forward_refs(self) -> bool: """ Verifies if the filed has any ForwardRefs that require updating before the model can be used. @@ -363,8 +327,7 @@ class BaseField(FieldInfo): """ return False - @classmethod - def evaluate_forward_ref(cls, globalns: Any, localns: Any) -> None: + def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None: """ Evaluates the ForwardRef to actual Field based on global and local namespaces @@ -376,8 +339,7 @@ class BaseField(FieldInfo): :rtype: None """ - @classmethod - def get_related_name(cls) -> str: + def get_related_name(self) -> str: """ Returns name to use for reverse relation. It's either set as `related_name` or by default it's owner model. get_name + 's' diff --git a/ormar/fields/foreign_key.py b/ormar/fields/foreign_key.py index 27ff23b..9458ce5 100644 --- a/ormar/fields/foreign_key.py +++ b/ormar/fields/foreign_key.py @@ -56,7 +56,7 @@ def create_dummy_instance(fk: Type["Model"], pk: Any = None) -> "Model": def create_dummy_model( base_model: Type["Model"], - pk_field: Type[Union[BaseField, "ForeignKeyField", "ManyToManyField"]], + pk_field: Union[BaseField, "ForeignKeyField", "ManyToManyField"], ) -> Type["BaseModel"]: """ Used to construct a dummy pydantic model for type hints and pydantic validation. @@ -65,7 +65,7 @@ def create_dummy_model( :param base_model: class of target dummy model :type base_model: Model class :param pk_field: ormar Field to be set on pydantic Model - :type pk_field: Type[Union[BaseField, "ForeignKeyField", "ManyToManyField"]] + :type pk_field: Union[BaseField, "ForeignKeyField", "ManyToManyField"] :return: constructed dummy model :rtype: pydantic.BaseModel """ @@ -256,7 +256,8 @@ def ForeignKey( # noqa CFQ002 related_orders_by=related_orders_by, ) - return type("ForeignKey", (ForeignKeyField, BaseField), namespace) + Field = type("ForeignKey", (ForeignKeyField, BaseField), {}) + return Field(**namespace) class ForeignKeyField(BaseField): @@ -264,15 +265,15 @@ class ForeignKeyField(BaseField): Actual class returned from ForeignKey function call and stored in model_fields. """ - to: Type["Model"] - name: str - related_name: str # type: ignore - virtual: bool - ondelete: str - onupdate: str + def __init__(self, **kwargs: Any) -> None: + if TYPE_CHECKING: # pragma: no cover + self.__type__: type + self.to: Type["Model"] + self.ondelete: str = kwargs.pop("ondelete", None) + self.onupdate: str = kwargs.pop("onupdate", None) + super().__init__(**kwargs) - @classmethod - def get_source_related_name(cls) -> str: + def get_source_related_name(self) -> str: """ Returns name to use for source relation name. For FK it's the same, differs for m2m fields. @@ -280,20 +281,18 @@ class ForeignKeyField(BaseField): :return: name of the related_name or default related name. :rtype: str """ - return cls.get_related_name() + return self.get_related_name() - @classmethod - def get_related_name(cls) -> str: + def get_related_name(self) -> str: """ Returns name to use for reverse relation. It's either set as `related_name` or by default it's owner model. get_name + 's' :return: name of the related_name or default related name. :rtype: str """ - return cls.related_name or cls.owner.get_name() + "s" + return self.related_name or self.owner.get_name() + "s" - @classmethod - def evaluate_forward_ref(cls, globalns: Any, localns: Any) -> None: + def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None: """ Evaluates the ForwardRef to actual Field based on global and local namespaces @@ -304,26 +303,25 @@ class ForeignKeyField(BaseField): :return: None :rtype: None """ - if cls.to.__class__ == ForwardRef: - cls.to = evaluate_forwardref( - cls.to, # type: ignore + if self.to.__class__ == ForwardRef: + self.to = evaluate_forwardref( + self.to, # type: ignore globalns, localns or None, ) ( - cls.__type__, - cls.constraints, - cls.column_type, + self.__type__, + self.constraints, + self.column_type, ) = populate_fk_params_based_on_to_model( - to=cls.to, - nullable=cls.nullable, - ondelete=cls.ondelete, - onupdate=cls.onupdate, + to=self.to, + nullable=self.nullable, + ondelete=self.ondelete, + onupdate=self.onupdate, ) - @classmethod def _extract_model_from_sequence( - cls, value: List, child: "Model", to_register: bool, + self, value: List, child: "Model", to_register: bool, ) -> List["Model"]: """ Takes a list of Models and registers them on parent. @@ -341,15 +339,14 @@ class ForeignKeyField(BaseField): :rtype: List["Model"] """ return [ - cls.expand_relationship( # type: ignore + self.expand_relationship( # type: ignore value=val, child=child, to_register=to_register, ) for val in value ] - @classmethod def _register_existing_model( - cls, value: "Model", child: "Model", to_register: bool, + self, value: "Model", child: "Model", to_register: bool, ) -> "Model": """ Takes already created instance and registers it for parent. @@ -367,12 +364,11 @@ class ForeignKeyField(BaseField): :rtype: Model """ if to_register: - cls.register_relation(model=value, child=child) + self.register_relation(model=value, child=child) return value - @classmethod def _construct_model_from_dict( - cls, value: dict, child: "Model", to_register: bool + self, value: dict, child: "Model", to_register: bool ) -> "Model": """ Takes a dictionary, creates a instance and registers it for parent. @@ -390,16 +386,15 @@ class ForeignKeyField(BaseField): :return: (if needed) registered Model :rtype: Model """ - if len(value.keys()) == 1 and list(value.keys())[0] == cls.to.Meta.pkname: + if len(value.keys()) == 1 and list(value.keys())[0] == self.to.Meta.pkname: value["__pk_only__"] = True - model = cls.to(**value) + model = self.to(**value) if to_register: - cls.register_relation(model=model, child=child) + self.register_relation(model=model, child=child) return model - @classmethod def _construct_model_from_pk( - cls, value: Any, child: "Model", to_register: bool + self, value: Any, child: "Model", to_register: bool ) -> "Model": """ Takes a pk value, creates a dummy instance and registers it for parent. @@ -416,21 +411,20 @@ class ForeignKeyField(BaseField): :return: (if needed) registered Model :rtype: Model """ - if cls.to.pk_type() == uuid.UUID and isinstance(value, str): # pragma: nocover + if self.to.pk_type() == uuid.UUID and isinstance(value, str): # pragma: nocover value = uuid.UUID(value) - if not isinstance(value, cls.to.pk_type()): + if not isinstance(value, self.to.pk_type()): raise RelationshipInstanceError( - f"Relationship error - ForeignKey {cls.to.__name__} " - f"is of type {cls.to.pk_type()} " + f"Relationship error - ForeignKey {self.to.__name__} " + f"is of type {self.to.pk_type()} " f"while {type(value)} passed as a parameter." ) - model = create_dummy_instance(fk=cls.to, pk=value) + model = create_dummy_instance(fk=self.to, pk=value) if to_register: - cls.register_relation(model=model, child=child) + self.register_relation(model=model, child=child) return model - @classmethod - def register_relation(cls, model: "Model", child: "Model") -> None: + def register_relation(self, model: "Model", child: "Model") -> None: """ Registers relation between parent and child in relation manager. Relation manager is kep on each model (different instance). @@ -444,11 +438,10 @@ class ForeignKeyField(BaseField): :type child: Model class """ model._orm.add( - parent=model, child=child, field=cls, + parent=model, child=child, field=self, ) - @classmethod - def has_unresolved_forward_refs(cls) -> bool: + def has_unresolved_forward_refs(self) -> bool: """ Verifies if the filed has any ForwardRefs that require updating before the model can be used. @@ -456,11 +449,10 @@ class ForeignKeyField(BaseField): :return: result of the check :rtype: bool """ - return cls.to.__class__ == ForwardRef + return self.to.__class__ == ForwardRef - @classmethod def expand_relationship( - cls, + self, value: Any, child: Union["Model", "NewBaseModel"], to_register: bool = True, @@ -483,20 +475,19 @@ class ForeignKeyField(BaseField): :rtype: Optional[Union["Model", List["Model"]]] """ if value is None: - return None if not cls.virtual else [] + return None if not self.virtual else [] constructors = { - f"{cls.to.__name__}": cls._register_existing_model, - "dict": cls._construct_model_from_dict, - "list": cls._extract_model_from_sequence, + f"{self.to.__name__}": self._register_existing_model, + "dict": self._construct_model_from_dict, + "list": self._extract_model_from_sequence, } model = constructors.get( # type: ignore - value.__class__.__name__, cls._construct_model_from_pk + value.__class__.__name__, self._construct_model_from_pk )(value, child, to_register) return model - @classmethod - def get_relation_name(cls) -> str: # pragma: no cover + def get_relation_name(self) -> str: # pragma: no cover """ Returns name of the relation, which can be a own name or through model names for m2m models @@ -504,14 +495,13 @@ class ForeignKeyField(BaseField): :return: result of the check :rtype: bool """ - return cls.name + return self.name - @classmethod - def get_source_model(cls) -> Type["Model"]: # pragma: no cover + def get_source_model(self) -> Type["Model"]: # pragma: no cover """ Returns model from which the relation comes -> either owner or through model :return: source model :rtype: Type["Model"] """ - return cls.owner + return self.owner diff --git a/ormar/fields/many_to_many.py b/ormar/fields/many_to_many.py index c7fb391..4f44121 100644 --- a/ormar/fields/many_to_many.py +++ b/ormar/fields/many_to_many.py @@ -132,7 +132,8 @@ def ManyToMany( related_orders_by=related_orders_by, ) - return type("ManyToMany", (ManyToManyField, BaseField), namespace) + Field = type("ManyToMany", (ManyToManyField, BaseField), {}) + return Field(**namespace) class ManyToManyField(ForeignKeyField, ormar.QuerySetProtocol, ormar.RelationProtocol): @@ -140,8 +141,14 @@ class ManyToManyField(ForeignKeyField, ormar.QuerySetProtocol, ormar.RelationPro Actual class returned from ManyToMany function call and stored in model_fields. """ - @classmethod - def get_source_related_name(cls) -> str: + def __init__(self, **kwargs: Any) -> None: + if TYPE_CHECKING: # pragma: no cover + self.__type__: type + self.to: Type["Model"] + self.through: Type["Model"] + super().__init__(**kwargs) + + def get_source_related_name(self) -> str: """ Returns name to use for source relation name. For FK it's the same, differs for m2m fields. @@ -150,32 +157,31 @@ class ManyToManyField(ForeignKeyField, ormar.QuerySetProtocol, ormar.RelationPro :rtype: str """ return ( - cls.through.Meta.model_fields[cls.default_source_field_name()].related_name - or cls.name + self.through.Meta.model_fields[ + self.default_source_field_name() + ].related_name + or self.name ) - @classmethod - def default_target_field_name(cls) -> str: + def default_target_field_name(self) -> str: """ Returns default target model name on through model. :return: name of the field :rtype: str """ - prefix = "from_" if cls.self_reference else "" - return f"{prefix}{cls.to.get_name()}" + prefix = "from_" if self.self_reference else "" + return f"{prefix}{self.to.get_name()}" - @classmethod - def default_source_field_name(cls) -> str: + def default_source_field_name(self) -> str: """ Returns default target model name on through model. :return: name of the field :rtype: str """ - prefix = "to_" if cls.self_reference else "" - return f"{prefix}{cls.owner.get_name()}" + prefix = "to_" if self.self_reference else "" + return f"{prefix}{self.owner.get_name()}" - @classmethod - def has_unresolved_forward_refs(cls) -> bool: + def has_unresolved_forward_refs(self) -> bool: """ Verifies if the filed has any ForwardRefs that require updating before the model can be used. @@ -183,10 +189,9 @@ class ManyToManyField(ForeignKeyField, ormar.QuerySetProtocol, ormar.RelationPro :return: result of the check :rtype: bool """ - return cls.to.__class__ == ForwardRef or cls.through.__class__ == ForwardRef + return self.to.__class__ == ForwardRef or self.through.__class__ == ForwardRef - @classmethod - def evaluate_forward_ref(cls, globalns: Any, localns: Any) -> None: + def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None: """ Evaluates the ForwardRef to actual Field based on global and local namespaces @@ -197,27 +202,26 @@ class ManyToManyField(ForeignKeyField, ormar.QuerySetProtocol, ormar.RelationPro :return: None :rtype: None """ - if cls.to.__class__ == ForwardRef: - cls.to = evaluate_forwardref( - cls.to, # type: ignore + if self.to.__class__ == ForwardRef: + self.to = evaluate_forwardref( + self.to, # type: ignore globalns, localns or None, ) - (cls.__type__, cls.column_type,) = populate_m2m_params_based_on_to_model( - to=cls.to, nullable=cls.nullable, + (self.__type__, self.column_type,) = populate_m2m_params_based_on_to_model( + to=self.to, nullable=self.nullable, ) - if cls.through.__class__ == ForwardRef: - cls.through = evaluate_forwardref( - cls.through, # type: ignore + if self.through.__class__ == ForwardRef: + self.through = evaluate_forwardref( + self.through, # type: ignore globalns, localns or None, ) - forbid_through_relations(cls.through) + forbid_through_relations(self.through) - @classmethod - def get_relation_name(cls) -> str: + def get_relation_name(self) -> str: """ Returns name of the relation, which can be a own name or through model names for m2m models @@ -225,34 +229,32 @@ class ManyToManyField(ForeignKeyField, ormar.QuerySetProtocol, ormar.RelationPro :return: result of the check :rtype: bool """ - if cls.self_reference and cls.name == cls.self_reference_primary: - return cls.default_source_field_name() - return cls.default_target_field_name() + if self.self_reference and self.name == self.self_reference_primary: + return self.default_source_field_name() + return self.default_target_field_name() - @classmethod - def get_source_model(cls) -> Type["Model"]: + def get_source_model(self) -> Type["Model"]: """ Returns model from which the relation comes -> either owner or through model :return: source model :rtype: Type["Model"] """ - return cls.through + return self.through - @classmethod - def create_default_through_model(cls) -> None: + def create_default_through_model(self) -> None: """ Creates default empty through model if no additional fields are required. """ - owner_name = cls.owner.get_name(lower=False) - to_name = cls.to.get_name(lower=False) + owner_name = self.owner.get_name(lower=False) + to_name = self.to.get_name(lower=False) class_name = f"{owner_name}{to_name}" table_name = f"{owner_name.lower()}s_{to_name.lower()}s" new_meta_namespace = { "tablename": table_name, - "database": cls.owner.Meta.database, - "metadata": cls.owner.Meta.metadata, + "database": self.owner.Meta.database, + "metadata": self.owner.Meta.metadata, } new_meta = type("Meta", (), new_meta_namespace) through_model = type(class_name, (ormar.Model,), {"Meta": new_meta}) - cls.through = cast(Type["Model"], through_model) + self.through = cast(Type["Model"], through_model) diff --git a/ormar/fields/model_fields.py b/ormar/fields/model_fields.py index 9cfd4f4..2236e71 100644 --- a/ormar/fields/model_fields.py +++ b/ormar/fields/model_fields.py @@ -1,7 +1,7 @@ import datetime import decimal import uuid -from typing import Any, Optional, TYPE_CHECKING, Type +from typing import Any, Optional, TYPE_CHECKING import pydantic import sqlalchemy @@ -63,7 +63,7 @@ class ModelFieldFactory: _bases: Any = (BaseField,) _type: Any = None - def __new__(cls, *args: Any, **kwargs: Any) -> Type[BaseField]: # type: ignore + def __new__(cls, *args: Any, **kwargs: Any) -> BaseField: # type: ignore cls.validate(**kwargs) default = kwargs.pop("default", None) @@ -77,7 +77,6 @@ class ModelFieldFactory: encrypt_secret = kwargs.pop("encrypt_secret", None) encrypt_backend = kwargs.pop("encrypt_backend", EncryptBackends.NONE) encrypt_custom_backend = kwargs.pop("encrypt_custom_backend", None) - encrypt_max_length = kwargs.pop("encrypt_max_length", 5000) namespace = dict( __type__=cls._type, @@ -97,10 +96,10 @@ class ModelFieldFactory: encrypt_secret=encrypt_secret, encrypt_backend=encrypt_backend, encrypt_custom_backend=encrypt_custom_backend, - encrypt_max_length=encrypt_max_length, **kwargs ) - return type(cls.__name__, cls._bases, namespace) + Field = type(cls.__name__, cls._bases, {}) + return Field(**namespace) @classmethod def get_column_type(cls, **kwargs: Any) -> Any: # pragma no cover @@ -141,7 +140,7 @@ class String(ModelFieldFactory, str): curtail_length: int = None, regex: str = None, **kwargs: Any - ) -> Type[BaseField]: # type: ignore + ) -> BaseField: # type: ignore kwargs = { **kwargs, **{ @@ -194,7 +193,7 @@ class Integer(ModelFieldFactory, int): maximum: int = None, multiple_of: int = None, **kwargs: Any - ) -> Type[BaseField]: + ) -> BaseField: autoincrement = kwargs.pop("autoincrement", None) autoincrement = ( autoincrement @@ -236,7 +235,7 @@ class Text(ModelFieldFactory, str): def __new__( # type: ignore cls, *, allow_blank: bool = True, strip_whitespace: bool = False, **kwargs: Any - ) -> Type[BaseField]: + ) -> BaseField: kwargs = { **kwargs, **{ @@ -276,7 +275,7 @@ class Float(ModelFieldFactory, float): maximum: float = None, multiple_of: int = None, **kwargs: Any - ) -> Type[BaseField]: + ) -> BaseField: kwargs = { **kwargs, **{ @@ -430,7 +429,7 @@ class BigInteger(Integer, int): maximum: int = None, multiple_of: int = None, **kwargs: Any - ) -> Type[BaseField]: + ) -> BaseField: autoincrement = kwargs.pop("autoincrement", None) autoincrement = ( autoincrement @@ -481,7 +480,7 @@ class Decimal(ModelFieldFactory, decimal.Decimal): max_digits: int = None, decimal_places: int = None, **kwargs: Any - ) -> Type[BaseField]: + ) -> BaseField: kwargs = { **kwargs, **{ @@ -544,7 +543,7 @@ class UUID(ModelFieldFactory, uuid.UUID): def __new__( # type: ignore # noqa CFQ002 cls, *, uuid_format: str = "hex", **kwargs: Any - ) -> Type[BaseField]: + ) -> BaseField: kwargs = { **kwargs, **{ diff --git a/ormar/fields/sqlalchemy_encrypted.py b/ormar/fields/sqlalchemy_encrypted.py index 1e6eda1..ff5ef29 100644 --- a/ormar/fields/sqlalchemy_encrypted.py +++ b/ormar/fields/sqlalchemy_encrypted.py @@ -133,7 +133,7 @@ class EncryptedString(types.TypeDecorator): raise ModelDefinitionError("Wrong or no encrypt backend provided!") self.backend: EncryptBackend = backend() - self._field_type: Type["BaseField"] = _field_type + self._field_type: "BaseField" = _field_type self._underlying_type: Any = _field_type.column_type self._key: Union[str, Callable] = encrypt_secret type_ = self._field_type.__type__ diff --git a/ormar/fields/through_field.py b/ormar/fields/through_field.py index b25e94b..9274b3b 100644 --- a/ormar/fields/through_field.py +++ b/ormar/fields/through_field.py @@ -57,7 +57,8 @@ def Through( # noqa CFQ002 is_through=True, ) - return type("Through", (ThroughField, BaseField), namespace) + Field = type("Through", (ThroughField, BaseField), {}) + return Field(**namespace) class ThroughField(ForeignKeyField): diff --git a/ormar/models/helpers/models.py b/ormar/models/helpers/models.py index 4ebe8dc..d94f8a8 100644 --- a/ormar/models/helpers/models.py +++ b/ormar/models/helpers/models.py @@ -12,7 +12,7 @@ if TYPE_CHECKING: # pragma no cover from ormar.fields import BaseField -def is_field_an_forward_ref(field: Type["BaseField"]) -> bool: +def is_field_an_forward_ref(field: "BaseField") -> bool: """ Checks if field is a relation field and whether any of the referenced models are ForwardRefs that needs to be updated before proceeding. diff --git a/ormar/models/helpers/pydantic.py b/ormar/models/helpers/pydantic.py index 60646f8..8c9e498 100644 --- a/ormar/models/helpers/pydantic.py +++ b/ormar/models/helpers/pydantic.py @@ -1,12 +1,10 @@ -import warnings from typing import Dict, Optional, TYPE_CHECKING, Tuple, Type import pydantic from pydantic.fields import ModelField from pydantic.utils import lenient_issubclass -import ormar # noqa: I100, I202 -from ormar.fields import BaseField +from ormar.fields import BaseField # noqa: I100, I202 if TYPE_CHECKING: # pragma no cover from ormar import Model @@ -14,7 +12,7 @@ if TYPE_CHECKING: # pragma no cover def create_pydantic_field( - field_name: str, model: Type["Model"], model_field: Type["ManyToManyField"] + field_name: str, model: Type["Model"], model_field: "ManyToManyField" ) -> None: """ Registers pydantic field on through model that leads to passed model @@ -59,38 +57,6 @@ def get_pydantic_field(field_name: str, model: Type["Model"]) -> "ModelField": ) -def populate_default_pydantic_field_value( - ormar_field: Type["BaseField"], field_name: str, attrs: dict -) -> dict: - """ - Grabs current value of the ormar Field in class namespace - (so the default_value declared on ormar model if set) - and converts it to pydantic.FieldInfo - that pydantic is able to extract later. - - On FieldInfo there are saved all needed params like max_length of the string - and other constraints that pydantic can use to build - it's own field validation used by ormar. - - :param ormar_field: field to convert - :type ormar_field: ormar Field - :param field_name: field to convert name - :type field_name: str - :param attrs: current class namespace - :type attrs: Dict - :return: updated namespace dict - :rtype: Dict - """ - curr_def_value = attrs.get(field_name, ormar.Undefined) - if lenient_issubclass(curr_def_value, ormar.fields.BaseField): - curr_def_value = ormar.Undefined - if curr_def_value is None: - attrs[field_name] = ormar_field.convert_to_pydantic_field_info(allow_null=True) - else: - attrs[field_name] = ormar_field.convert_to_pydantic_field_info() - return attrs - - def populate_pydantic_default_values(attrs: Dict) -> Tuple[Dict, Dict]: """ Extracts ormar fields from annotations (deprecated) and from namespace @@ -110,22 +76,11 @@ def populate_pydantic_default_values(attrs: Dict) -> Tuple[Dict, Dict]: :rtype: Tuple[Dict, Dict] """ model_fields = {} - potential_fields = { - k: v - for k, v in attrs["__annotations__"].items() - if lenient_issubclass(v, BaseField) - } - if potential_fields: - warnings.warn( - "Using ormar.Fields as type Model annotation has been deprecated," - " check documentation of current version", - DeprecationWarning, - ) + potential_fields = {} potential_fields.update(get_potential_fields(attrs)) for field_name, field in potential_fields.items(): field.name = field_name - attrs = populate_default_pydantic_field_value(field, field_name, attrs) model_fields[field_name] = field attrs["__annotations__"][field_name] = ( field.__type__ if not field.nullable else Optional[field.__type__] @@ -156,4 +111,8 @@ def get_potential_fields(attrs: Dict) -> Dict: :return: extracted fields that are ormar Fields :rtype: Dict """ - return {k: v for k, v in attrs.items() if lenient_issubclass(v, BaseField)} + return { + k: v + for k, v in attrs.items() + if (lenient_issubclass(v, BaseField) or isinstance(v, BaseField)) + } diff --git a/ormar/models/helpers/relations.py b/ormar/models/helpers/relations.py index 6dc77b9..ac35d9a 100644 --- a/ormar/models/helpers/relations.py +++ b/ormar/models/helpers/relations.py @@ -13,7 +13,7 @@ if TYPE_CHECKING: # pragma no cover alias_manager = AliasManager() -def register_relation_on_build(field: Type["ForeignKeyField"]) -> None: +def register_relation_on_build(field: "ForeignKeyField") -> None: """ Registers ForeignKey relation in alias_manager to set a table_prefix. Registration include also reverse relation side to be able to join both sides. @@ -32,7 +32,7 @@ def register_relation_on_build(field: Type["ForeignKeyField"]) -> None: ) -def register_many_to_many_relation_on_build(field: Type["ManyToManyField"]) -> None: +def register_many_to_many_relation_on_build(field: "ManyToManyField") -> None: """ Registers connection between through model and both sides of the m2m relation. Registration include also reverse relation side to be able to join both sides. @@ -58,7 +58,7 @@ def register_many_to_many_relation_on_build(field: Type["ManyToManyField"]) -> N ) -def expand_reverse_relationship(model_field: Type["ForeignKeyField"]) -> None: +def expand_reverse_relationship(model_field: "ForeignKeyField") -> None: """ If the reverse relation has not been set before it's set here. @@ -84,11 +84,11 @@ def expand_reverse_relationships(model: Type["Model"]) -> None: model_fields = list(model.Meta.model_fields.values()) for model_field in model_fields: if model_field.is_relation and not model_field.has_unresolved_forward_refs(): - model_field = cast(Type["ForeignKeyField"], model_field) + model_field = cast("ForeignKeyField", model_field) expand_reverse_relationship(model_field=model_field) -def register_reverse_model_fields(model_field: Type["ForeignKeyField"]) -> None: +def register_reverse_model_fields(model_field: "ForeignKeyField") -> None: """ Registers reverse ForeignKey field on related model. By default it's name.lower()+'s' of the model on which relation is defined. @@ -113,7 +113,7 @@ def register_reverse_model_fields(model_field: Type["ForeignKeyField"]) -> None: orders_by=model_field.related_orders_by, ) # register foreign keys on through model - model_field = cast(Type["ManyToManyField"], model_field) + model_field = cast("ManyToManyField", model_field) register_through_shortcut_fields(model_field=model_field) adjust_through_many_to_many_model(model_field=model_field) else: @@ -128,7 +128,7 @@ def register_reverse_model_fields(model_field: Type["ForeignKeyField"]) -> None: ) -def register_through_shortcut_fields(model_field: Type["ManyToManyField"]) -> None: +def register_through_shortcut_fields(model_field: "ManyToManyField") -> None: """ Registers m2m relation through shortcut on both ends of the relation. @@ -156,7 +156,7 @@ def register_through_shortcut_fields(model_field: Type["ManyToManyField"]) -> No ) -def register_relation_in_alias_manager(field: Type["ForeignKeyField"]) -> None: +def register_relation_in_alias_manager(field: "ForeignKeyField") -> None: """ Registers the relation (and reverse relation) in alias manager. The m2m relations require registration of through model between @@ -172,7 +172,7 @@ def register_relation_in_alias_manager(field: Type["ForeignKeyField"]) -> None: if field.is_multi: if field.has_unresolved_forward_refs(): return - field = cast(Type["ManyToManyField"], field) + field = cast("ManyToManyField", field) register_many_to_many_relation_on_build(field=field) elif field.is_relation and not field.is_through: if field.has_unresolved_forward_refs(): @@ -181,7 +181,7 @@ def register_relation_in_alias_manager(field: Type["ForeignKeyField"]) -> None: def verify_related_name_dont_duplicate( - related_name: str, model_field: Type["ForeignKeyField"] + related_name: str, model_field: "ForeignKeyField" ) -> None: """ Verifies whether the used related_name (regardless of the fact if user defined or @@ -213,7 +213,7 @@ def verify_related_name_dont_duplicate( ) -def reverse_field_not_already_registered(model_field: Type["ForeignKeyField"]) -> bool: +def reverse_field_not_already_registered(model_field: "ForeignKeyField") -> bool: """ Checks if child is already registered in parents pydantic fields. diff --git a/ormar/models/helpers/sqlalchemy.py b/ormar/models/helpers/sqlalchemy.py index 905af75..fdb23c2 100644 --- a/ormar/models/helpers/sqlalchemy.py +++ b/ormar/models/helpers/sqlalchemy.py @@ -14,7 +14,7 @@ if TYPE_CHECKING: # pragma no cover from ormar.models import NewBaseModel -def adjust_through_many_to_many_model(model_field: Type["ManyToManyField"]) -> None: +def adjust_through_many_to_many_model(model_field: "ManyToManyField") -> None: """ Registers m2m relation on through model. Sets ormar.ForeignKey from through model to both child and parent models. @@ -52,7 +52,7 @@ def adjust_through_many_to_many_model(model_field: Type["ManyToManyField"]) -> N def create_and_append_m2m_fk( - model: Type["Model"], model_field: Type["ManyToManyField"], field_name: str + model: Type["Model"], model_field: "ManyToManyField", field_name: str ) -> None: """ Registers sqlalchemy Column with sqlalchemy.ForeignKey leading to the model. @@ -190,22 +190,22 @@ def _process_fields( return pkname, columns -def _is_through_model_not_set(field: Type["BaseField"]) -> bool: +def _is_through_model_not_set(field: "BaseField") -> bool: """ Alias to if check that verifies if through model was created. :param field: field to check - :type field: Type["BaseField"] + :type field: "BaseField" :return: result of the check :rtype: bool """ return field.is_multi and not field.through -def _is_db_field(field: Type["BaseField"]) -> bool: +def _is_db_field(field: "BaseField") -> bool: """ Alias to if check that verifies if field should be included in database. :param field: field to check - :type field: Type["BaseField"] + :type field: "BaseField" :return: result of the check :rtype: bool """ @@ -298,7 +298,7 @@ def populate_meta_sqlalchemy_table_if_required(meta: "ModelMeta") -> None: def update_column_definition( - model: Union[Type["Model"], Type["NewBaseModel"]], field: Type["ForeignKeyField"] + model: Union[Type["Model"], Type["NewBaseModel"]], field: "ForeignKeyField" ) -> None: """ Updates a column with a new type column based on updated parameters in FK fields. @@ -306,7 +306,7 @@ def update_column_definition( :param model: model on which columns needs to be updated :type model: Type["Model"] :param field: field with column definition that requires update - :type field: Type[ForeignKeyField] + :type field: ForeignKeyField :return: None :rtype: None """ diff --git a/ormar/models/helpers/validation.py b/ormar/models/helpers/validation.py index bc87235..a535f65 100644 --- a/ormar/models/helpers/validation.py +++ b/ormar/models/helpers/validation.py @@ -20,7 +20,7 @@ if TYPE_CHECKING: # pragma no cover from ormar import Model -def check_if_field_has_choices(field: Type[BaseField]) -> bool: +def check_if_field_has_choices(field: BaseField) -> bool: """ Checks if given field has choices populated. A if it has one, a validator for this field needs to be attached. @@ -34,7 +34,7 @@ def check_if_field_has_choices(field: Type[BaseField]) -> bool: def convert_choices_if_needed( # noqa: CCR001 - field: Type["BaseField"], value: Any + field: "BaseField", value: Any ) -> Tuple[Any, List]: """ Converts dates to isoformat as fastapi can check this condition in routes @@ -47,7 +47,7 @@ def convert_choices_if_needed( # noqa: CCR001 Converts decimal to float with given scale. :param field: ormar field to check with choices - :type field: Type[BaseField] + :type field: BaseField :param values: current values of the model to verify :type values: Dict :return: value, choices list @@ -77,13 +77,13 @@ def convert_choices_if_needed( # noqa: CCR001 return value, choices -def validate_choices(field: Type["BaseField"], value: Any) -> None: +def validate_choices(field: "BaseField", value: Any) -> None: """ Validates if given value is in provided choices. :raises ValueError: If value is not in choices. :param field:field to validate - :type field: Type[BaseField] + :type field: BaseField :param value: value of the field :type value: Any """ diff --git a/ormar/models/metaclass.py b/ormar/models/metaclass.py index 116a592..6c529f2 100644 --- a/ormar/models/metaclass.py +++ b/ormar/models/metaclass.py @@ -63,9 +63,7 @@ class ModelMeta: columns: List[sqlalchemy.Column] constraints: List[ColumnCollectionConstraint] pkname: str - model_fields: Dict[ - str, Union[Type[BaseField], Type[ForeignKeyField], Type[ManyToManyField]] - ] + model_fields: Dict[str, Union[BaseField, ForeignKeyField, ManyToManyField]] alias_manager: AliasManager property_fields: Set signals: SignalEmitter @@ -215,7 +213,7 @@ def update_attrs_from_base_meta( # noqa: CCR001 def copy_and_replace_m2m_through_model( # noqa: CFQ002 - field: Type[ManyToManyField], + field: ManyToManyField, field_name: str, table_name: str, parent_fields: Dict, @@ -238,7 +236,7 @@ def copy_and_replace_m2m_through_model( # noqa: CFQ002 :param base_class: base class model :type base_class: Type["Model"] :param field: field with relations definition - :type field: Type[ManyToManyField] + :type field: ManyToManyField :param field_name: name of the relation field :type field_name: str :param table_name: name of the table @@ -250,9 +248,10 @@ def copy_and_replace_m2m_through_model( # noqa: CFQ002 :param meta: metaclass of currently created model :type meta: ModelMeta """ - copy_field: Type[BaseField] = type( # type: ignore - field.__name__, (ManyToManyField, BaseField), dict(field.__dict__) + Field: Type[BaseField] = type( # type: ignore + field.__class__.__name__, (ManyToManyField, BaseField), {} ) + copy_field = Field(**dict(field.__dict__)) related_name = field.related_name + "_" + table_name copy_field.related_name = related_name # type: ignore @@ -293,9 +292,7 @@ def copy_data_from_parent_model( # noqa: CCR001 base_class: Type["Model"], curr_class: type, attrs: Dict, - model_fields: Dict[ - str, Union[Type[BaseField], Type[ForeignKeyField], Type[ManyToManyField]] - ], + model_fields: Dict[str, Union[BaseField, ForeignKeyField, ManyToManyField]], ) -> Tuple[Dict, Dict]: """ Copy the key parameters [databse, metadata, property_fields and constraints] @@ -342,7 +339,7 @@ def copy_data_from_parent_model( # noqa: CCR001 ) for field_name, field in base_class.Meta.model_fields.items(): if field.is_multi: - field = cast(Type["ManyToManyField"], field) + field = cast(ManyToManyField, field) copy_and_replace_m2m_through_model( field=field, field_name=field_name, @@ -354,9 +351,10 @@ def copy_data_from_parent_model( # noqa: CCR001 ) elif field.is_relation and field.related_name: - copy_field = type( # type: ignore - field.__name__, (ForeignKeyField, BaseField), dict(field.__dict__) + Field = type( # type: ignore + field.__class__.__name__, (ForeignKeyField, BaseField), {} ) + copy_field = Field(**dict(field.__dict__)) related_name = field.related_name + "_" + table_name copy_field.related_name = related_name # type: ignore parent_fields[field_name] = copy_field @@ -372,9 +370,7 @@ def extract_from_parents_definition( # noqa: CCR001 base_class: type, curr_class: type, attrs: Dict, - model_fields: Dict[ - str, Union[Type[BaseField], Type[ForeignKeyField], Type[ManyToManyField]] - ], + model_fields: Dict[str, Union[BaseField, ForeignKeyField, ManyToManyField]], ) -> Tuple[Dict, Dict]: """ Extracts fields from base classes if they have valid oramr fields. diff --git a/ormar/models/mixins/alias_mixin.py b/ormar/models/mixins/alias_mixin.py index 2108554..2d98657 100644 --- a/ormar/models/mixins/alias_mixin.py +++ b/ormar/models/mixins/alias_mixin.py @@ -67,6 +67,6 @@ class AliasMixin: :rtype: Dict """ for field_name, field in cls.Meta.model_fields.items(): - if field.alias and field.alias in new_kwargs: - new_kwargs[field_name] = new_kwargs.pop(field.alias) + if field.get_alias() and field.get_alias() in new_kwargs: + new_kwargs[field_name] = new_kwargs.pop(field.get_alias()) return new_kwargs diff --git a/ormar/models/mixins/prefetch_mixin.py b/ormar/models/mixins/prefetch_mixin.py index 85faec2..6720575 100644 --- a/ormar/models/mixins/prefetch_mixin.py +++ b/ormar/models/mixins/prefetch_mixin.py @@ -41,7 +41,7 @@ class PrefetchQueryMixin(RelationMixin): field_name = parent_model.Meta.model_fields[related].get_related_name() field = target_model.Meta.model_fields[field_name] if field.is_multi: - field = cast(Type["ManyToManyField"], field) + field = cast("ManyToManyField", field) field_name = field.default_target_field_name() sub_field = field.through.Meta.model_fields[field_name] return field.through, sub_field.get_alias() @@ -78,7 +78,7 @@ class PrefetchQueryMixin(RelationMixin): return column.get_alias() if use_raw else column.name @classmethod - def get_related_field_name(cls, target_field: Type["ForeignKeyField"]) -> str: + def get_related_field_name(cls, target_field: "ForeignKeyField") -> str: """ Returns name of the relation field that should be used in prefetch query. This field is later used to register relation in prefetch query, diff --git a/ormar/models/mixins/relation_mixin.py b/ormar/models/mixins/relation_mixin.py index 8955acf..2a20dcc 100644 --- a/ormar/models/mixins/relation_mixin.py +++ b/ormar/models/mixins/relation_mixin.py @@ -1,4 +1,3 @@ -import inspect from typing import ( Callable, List, @@ -9,6 +8,8 @@ from typing import ( Union, ) +from ormar import BaseField + class RelationMixin: """ @@ -85,7 +86,11 @@ class RelationMixin: related_names = set() for name, field in cls.Meta.model_fields.items(): - if inspect.isclass(field) and field.is_relation and not field.is_through: + if ( + isinstance(field, BaseField) + and field.is_relation + and not field.is_through + ): related_names.add(name) cls._related_names = related_names diff --git a/ormar/models/model_row.py b/ormar/models/model_row.py index d06d0d9..461cf9b 100644 --- a/ormar/models/model_row.py +++ b/ormar/models/model_row.py @@ -29,7 +29,7 @@ class ModelRow(NewBaseModel): source_model: Type["Model"], select_related: List = None, related_models: Any = None, - related_field: Type["ForeignKeyField"] = None, + related_field: "ForeignKeyField" = None, excludable: ExcludableItems = None, current_relation_str: str = "", proxy_source_model: Optional[Type["Model"]] = None, @@ -65,7 +65,7 @@ class ModelRow(NewBaseModel): :param related_models: list or dict of related models :type related_models: Union[List, Dict] :param related_field: field with relation declaration - :type related_field: Type[ForeignKeyField] + :type related_field: ForeignKeyField :return: returns model if model is populated from database :rtype: Optional[Model] """ @@ -116,7 +116,7 @@ class ModelRow(NewBaseModel): cls, source_model: Type["Model"], current_relation_str: str, - related_field: Type["ForeignKeyField"], + related_field: "ForeignKeyField", used_prefixes: List[str], ) -> str: """ @@ -126,7 +126,7 @@ class ModelRow(NewBaseModel): :param current_relation_str: current relation string :type current_relation_str: str :param related_field: field with relation declaration - :type related_field: Type["ForeignKeyField"] + :type related_field: "ForeignKeyField" :param used_prefixes: list of already extracted prefixes :type used_prefixes: List[str] :return: table_prefix to use @@ -193,7 +193,7 @@ class ModelRow(NewBaseModel): for related in related_models: field = cls.Meta.model_fields[related] - field = cast(Type["ForeignKeyField"], field) + field = cast("ForeignKeyField", field) model_cls = field.to model_excludable = excludable.get( model_cls=cast(Type["Model"], cls), alias=table_prefix diff --git a/ormar/models/newbasemodel.py b/ormar/models/newbasemodel.py index 105aa87..02a2663 100644 --- a/ormar/models/newbasemodel.py +++ b/ormar/models/newbasemodel.py @@ -67,7 +67,7 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass __slots__ = ("_orm_id", "_orm_saved", "_orm", "_pk_column") if TYPE_CHECKING: # pragma no cover - __model_fields__: Dict[str, Type[BaseField]] + __model_fields__: Dict[str, BaseField] __table__: sqlalchemy.Table __fields__: Dict[str, pydantic.fields.ModelField] __pydantic_model__: Type[BaseModel] @@ -455,7 +455,7 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass fields_to_check = cls.Meta.model_fields.copy() for field in fields_to_check.values(): if field.has_unresolved_forward_refs(): - field = cast(Type[ForeignKeyField], field) + field = cast(ForeignKeyField, field) field.evaluate_forward_ref(globalns=globalns, localns=localns) field.set_self_reference_flag() expand_reverse_relationship(model_field=field) @@ -747,12 +747,12 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass ) return self_fields - def get_relation_model_id(self, target_field: Type["BaseField"]) -> Optional[int]: + def get_relation_model_id(self, target_field: "BaseField") -> Optional[int]: """ Returns an id of the relation side model to use in prefetch query. :param target_field: field with relation definition - :type target_field: Type["BaseField"] + :type target_field: "BaseField" :return: value of pk if set :rtype: Optional[int] """ diff --git a/ormar/queryset/prefetch_query.py b/ormar/queryset/prefetch_query.py index 7d53535..e768b1b 100644 --- a/ormar/queryset/prefetch_query.py +++ b/ormar/queryset/prefetch_query.py @@ -292,7 +292,7 @@ class PrefetchQuery: for related in related_to_extract: target_field = model.Meta.model_fields[related] - target_field = cast(Type["ForeignKeyField"], target_field) + target_field = cast("ForeignKeyField", target_field) target_model = target_field.to.get_name() model_id = model.get_relation_model_id(target_field=target_field) @@ -394,7 +394,7 @@ class PrefetchQuery: :rtype: None """ target_field = target_model.Meta.model_fields[related] - target_field = cast(Type["ForeignKeyField"], target_field) + target_field = cast("ForeignKeyField", target_field) reverse = False if target_field.virtual or target_field.is_multi: reverse = True @@ -461,7 +461,7 @@ class PrefetchQuery: async def _run_prefetch_query( self, - target_field: Type["BaseField"], + target_field: "BaseField", excludable: "ExcludableItems", filter_clauses: List, related_field_name: str, @@ -474,7 +474,7 @@ class PrefetchQuery: models. :param target_field: ormar field with relation definition - :type target_field: Type["BaseField"] + :type target_field: "BaseField" :param filter_clauses: list of clauses, actually one clause with ids of relation :type filter_clauses: List[sqlalchemy.sql.elements.TextClause] :return: table prefix and raw rows from sql response @@ -540,13 +540,13 @@ class PrefetchQuery: ) def _update_already_loaded_rows( # noqa: CFQ002 - self, target_field: Type["BaseField"], prefetch_dict: Dict, orders_by: Dict, + self, target_field: "BaseField", prefetch_dict: Dict, orders_by: Dict, ) -> None: """ Updates models that are already loaded, usually children of children. :param target_field: ormar field with relation definition - :type target_field: Type["BaseField"] + :type target_field: "BaseField" :param prefetch_dict: dictionaries of related models to prefetch :type prefetch_dict: Dict :param orders_by: dictionary of order by clauses by model @@ -561,7 +561,7 @@ class PrefetchQuery: def _populate_rows( # noqa: CFQ002 self, rows: List, - target_field: Type["ForeignKeyField"], + target_field: "ForeignKeyField", parent_model: Type["Model"], table_prefix: str, exclude_prefix: str, @@ -584,7 +584,7 @@ class PrefetchQuery: :param rows: raw sql response from the prefetch query :type rows: List[sqlalchemy.engine.result.RowProxy] :param target_field: field with relation definition from parent model - :type target_field: Type["BaseField"] + :type target_field: "BaseField" :param parent_model: model with relation definition :type parent_model: Type[Model] :param table_prefix: prefix of the target table from current relation diff --git a/ormar/queryset/utils.py b/ormar/queryset/utils.py index 64e8af2..f5f00ac 100644 --- a/ormar/queryset/utils.py +++ b/ormar/queryset/utils.py @@ -264,7 +264,7 @@ def get_relationship_alias_model_and_str( def _process_through_field( related_parts: List, relation: Optional[str], - related_field: Type["BaseField"], + related_field: "BaseField", previous_model: Type["Model"], previous_models: List[Type["Model"]], ) -> Tuple[Type["Model"], Optional[str], bool]: @@ -276,7 +276,7 @@ def _process_through_field( :param relation: relation name :type relation: str :param related_field: field with relation declaration - :type related_field: Type["ForeignKeyField"] + :type related_field: "ForeignKeyField" :param previous_model: model from which relation is coming :type previous_model: Type["Model"] :param previous_models: list of already visited models in relation chain diff --git a/ormar/relations/alias_manager.py b/ormar/relations/alias_manager.py index 2ec6159..adb978d 100644 --- a/ormar/relations/alias_manager.py +++ b/ormar/relations/alias_manager.py @@ -151,7 +151,7 @@ class AliasManager: self, source_model: Union[Type["Model"], Type["ModelRow"]], relation_str: str, - relation_field: Type["ForeignKeyField"], + relation_field: "ForeignKeyField", ) -> str: """ Given source model and relation string returns the alias for this complex @@ -159,7 +159,7 @@ class AliasManager: field definition. :param relation_field: field with direct relation definition - :type relation_field: Type["ForeignKeyField"] + :type relation_field: "ForeignKeyField" :param source_model: model with query starts :type source_model: source Model :param relation_str: string with relation joins defined diff --git a/ormar/relations/relation_manager.py b/ormar/relations/relation_manager.py index 19c0dc5..d5295ec 100644 --- a/ormar/relations/relation_manager.py +++ b/ormar/relations/relation_manager.py @@ -16,7 +16,7 @@ class RelationsManager: def __init__( self, - related_fields: List[Type["ForeignKeyField"]] = None, + related_fields: List["ForeignKeyField"] = None, owner: Optional["Model"] = None, ) -> None: self.owner = proxy(owner) @@ -57,7 +57,7 @@ class RelationsManager: return None # pragma nocover @staticmethod - def add(parent: "Model", child: "Model", field: Type["ForeignKeyField"],) -> None: + def add(parent: "Model", child: "Model", field: "ForeignKeyField",) -> None: """ Adds relation on both sides -> meaning on both child and parent models. One side of the relation is always weakref proxy to avoid circular refs. @@ -138,12 +138,12 @@ class RelationsManager: return relation return None - def _get_relation_type(self, field: Type["BaseField"]) -> RelationType: + def _get_relation_type(self, field: "BaseField") -> RelationType: """ Returns type of the relation declared on a field. :param field: field with relation declaration - :type field: Type[BaseField] + :type field: BaseField :return: type of the relation defined on field :rtype: RelationType """ @@ -153,13 +153,13 @@ class RelationsManager: return RelationType.THROUGH return RelationType.PRIMARY if not field.virtual else RelationType.REVERSE - def _add_relation(self, field: Type["BaseField"]) -> None: + def _add_relation(self, field: "BaseField") -> None: """ Registers relation in the manager. Adds Relation instance under field.name. :param field: field with relation declaration - :type field: Type[BaseField] + :type field: BaseField """ self._relations[field.name] = Relation( manager=self, diff --git a/ormar/relations/utils.py b/ormar/relations/utils.py index f9315f2..585ff91 100644 --- a/ormar/relations/utils.py +++ b/ormar/relations/utils.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Tuple, Type +from typing import TYPE_CHECKING, Tuple from weakref import proxy from ormar.fields.foreign_key import ForeignKeyField @@ -8,7 +8,7 @@ if TYPE_CHECKING: # pragma no cover def get_relations_sides_and_names( - to_field: Type[ForeignKeyField], parent: "Model", child: "Model", + to_field: ForeignKeyField, parent: "Model", child: "Model", ) -> Tuple["Model", "Model", str, str]: """ Determines the names of child and parent relations names, as well as diff --git a/tests/test_inheritance_concrete.py b/tests/test_inheritance_concrete.py index 6bc3859..ac059e4 100644 --- a/tests/test_inheritance_concrete.py +++ b/tests/test_inheritance_concrete.py @@ -193,8 +193,8 @@ def test_field_redefining_in_concrete_models(): created_date: str = ormar.String(max_length=200, name="creation_date") changed_field = RedefinedField.Meta.model_fields["created_date"] - assert changed_field.default is None - assert changed_field.alias == "creation_date" + assert changed_field.ormar_default is None + assert changed_field.get_alias() == "creation_date" assert any(x.name == "creation_date" for x in RedefinedField.Meta.table.columns) assert isinstance( RedefinedField.Meta.table.columns["creation_date"].type, sa.sql.sqltypes.String, diff --git a/tests/test_inheritance_mixins.py b/tests/test_inheritance_mixins.py index bb423f1..6e3b565 100644 --- a/tests/test_inheritance_mixins.py +++ b/tests/test_inheritance_mixins.py @@ -64,8 +64,10 @@ def test_field_redefining(): id: int = ormar.Integer(primary_key=True) created_date: datetime.datetime = ormar.DateTime(name="creation_date") - assert RedefinedField.Meta.model_fields["created_date"].default is None - assert RedefinedField.Meta.model_fields["created_date"].alias == "creation_date" + assert RedefinedField.Meta.model_fields["created_date"].ormar_default is None + assert ( + RedefinedField.Meta.model_fields["created_date"].get_alias() == "creation_date" + ) assert any(x.name == "creation_date" for x in RedefinedField.Meta.table.columns) @@ -87,8 +89,10 @@ def test_field_redefining_in_second_raises_error(): id: int = ormar.Integer(primary_key=True) created_date: str = ormar.String(max_length=200, name="creation_date") - assert RedefinedField2.Meta.model_fields["created_date"].default is None - assert RedefinedField2.Meta.model_fields["created_date"].alias == "creation_date" + assert RedefinedField2.Meta.model_fields["created_date"].ormar_default is None + assert ( + RedefinedField2.Meta.model_fields["created_date"].get_alias() == "creation_date" + ) assert any(x.name == "creation_date" for x in RedefinedField2.Meta.table.columns) assert isinstance( RedefinedField2.Meta.table.columns["creation_date"].type, diff --git a/tests/test_model_definition.py b/tests/test_model_definition.py index bac267c..f3cf30f 100644 --- a/tests/test_model_definition.py +++ b/tests/test_model_definition.py @@ -29,7 +29,7 @@ class ExampleModel(Model): test_string: str = ormar.String(max_length=250) test_text: str = ormar.Text(default="") test_bool: bool = ormar.Boolean(nullable=False) - test_float: ormar.Float() = None # type: ignore + test_float = ormar.Float(nullable=True) test_datetime = ormar.DateTime(default=datetime.datetime.now) test_date = ormar.Date(default=datetime.date.today) test_time = ormar.Time(default=datetime.time) diff --git a/tests/test_models.py b/tests/test_models.py index 2724eec..b15ec84 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -120,9 +120,9 @@ async def create_test_database(): def test_model_class(): assert list(User.Meta.model_fields.keys()) == ["id", "name"] - assert issubclass(User.Meta.model_fields["id"], pydantic.fields.FieldInfo) + assert issubclass(User.Meta.model_fields["id"].__class__, pydantic.fields.FieldInfo) assert User.Meta.model_fields["id"].primary_key is True - assert issubclass(User.Meta.model_fields["name"], pydantic.fields.FieldInfo) + assert isinstance(User.Meta.model_fields["name"], pydantic.fields.FieldInfo) assert User.Meta.model_fields["name"].max_length == 100 assert isinstance(User.Meta.table, sqlalchemy.Table)