version with pydantic inheritance passing all the tests
This commit is contained in:
collerek
37
ormar/__init__.py
Normal file
37
ormar/__init__.py
Normal file
@ -0,0 +1,37 @@
|
||||
from ormar.exceptions import ModelDefinitionError, ModelNotSet, MultipleMatches, NoMatch
|
||||
from ormar.fields import (
|
||||
BigInteger,
|
||||
Boolean,
|
||||
Date,
|
||||
DateTime,
|
||||
Decimal,
|
||||
Float,
|
||||
ForeignKey,
|
||||
Integer,
|
||||
JSON,
|
||||
String,
|
||||
Text,
|
||||
Time,
|
||||
)
|
||||
from ormar.models import Model
|
||||
|
||||
__version__ = "0.1.3"
|
||||
__all__ = [
|
||||
"Integer",
|
||||
"BigInteger",
|
||||
"Boolean",
|
||||
"Time",
|
||||
"Text",
|
||||
"String",
|
||||
"JSON",
|
||||
"DateTime",
|
||||
"Date",
|
||||
"Decimal",
|
||||
"Float",
|
||||
"Model",
|
||||
"ModelDefinitionError",
|
||||
"ModelNotSet",
|
||||
"MultipleMatches",
|
||||
"NoMatch",
|
||||
"ForeignKey",
|
||||
]
|
||||
26
ormar/exceptions.py
Normal file
26
ormar/exceptions.py
Normal file
@ -0,0 +1,26 @@
|
||||
class AsyncOrmException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class ModelDefinitionError(AsyncOrmException):
|
||||
pass
|
||||
|
||||
|
||||
class ModelNotSet(AsyncOrmException):
|
||||
pass
|
||||
|
||||
|
||||
class NoMatch(AsyncOrmException):
|
||||
pass
|
||||
|
||||
|
||||
class MultipleMatches(AsyncOrmException):
|
||||
pass
|
||||
|
||||
|
||||
class QueryDefinitionError(AsyncOrmException):
|
||||
pass
|
||||
|
||||
|
||||
class RelationshipInstanceError(AsyncOrmException):
|
||||
pass
|
||||
31
ormar/fields/__init__.py
Normal file
31
ormar/fields/__init__.py
Normal file
@ -0,0 +1,31 @@
|
||||
from ormar.fields.base import BaseField
|
||||
from ormar.fields.foreign_key import ForeignKey
|
||||
from ormar.fields.model_fields import (
|
||||
BigInteger,
|
||||
Boolean,
|
||||
Date,
|
||||
DateTime,
|
||||
Decimal,
|
||||
Float,
|
||||
Integer,
|
||||
JSON,
|
||||
String,
|
||||
Text,
|
||||
Time,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"Decimal",
|
||||
"BigInteger",
|
||||
"Boolean",
|
||||
"Date",
|
||||
"DateTime",
|
||||
"String",
|
||||
"JSON",
|
||||
"Integer",
|
||||
"Text",
|
||||
"Float",
|
||||
"Time",
|
||||
"ForeignKey",
|
||||
"BaseField",
|
||||
]
|
||||
80
ormar/fields/base.py
Normal file
80
ormar/fields/base.py
Normal file
@ -0,0 +1,80 @@
|
||||
from typing import Any, Dict, List, Optional, TYPE_CHECKING
|
||||
|
||||
import pydantic
|
||||
import sqlalchemy
|
||||
from pydantic import Field
|
||||
|
||||
from ormar import ModelDefinitionError # noqa I101
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar.models import Model
|
||||
|
||||
|
||||
def prepare_validator(type_):
|
||||
def validate_model_field(value):
|
||||
return isinstance(value, type_)
|
||||
|
||||
return validate_model_field
|
||||
|
||||
|
||||
class BaseField:
|
||||
__type__ = None
|
||||
|
||||
column_type: sqlalchemy.Column
|
||||
constraints: List = []
|
||||
|
||||
primary_key: bool
|
||||
autoincrement: bool
|
||||
nullable: bool
|
||||
index: bool
|
||||
unique: bool
|
||||
pydantic_only: bool
|
||||
|
||||
default: Any
|
||||
server_default: Any
|
||||
|
||||
@classmethod
|
||||
def is_required(cls) -> bool:
|
||||
return (
|
||||
not cls.nullable and not cls.has_default() and not cls.is_auto_primary_key()
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def default_value(cls):
|
||||
if cls.is_auto_primary_key():
|
||||
return Field(default=None)
|
||||
if cls.has_default():
|
||||
default = cls.default if cls.default is not None else cls.server_default
|
||||
if callable(default):
|
||||
return Field(default_factory=default)
|
||||
else:
|
||||
return Field(default=default)
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def has_default(cls):
|
||||
return cls.default is not None or cls.server_default is not None
|
||||
|
||||
@classmethod
|
||||
def is_auto_primary_key(cls) -> bool:
|
||||
if cls.primary_key:
|
||||
return cls.autoincrement
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def get_column(cls, name: str) -> sqlalchemy.Column:
|
||||
return sqlalchemy.Column(
|
||||
name,
|
||||
cls.column_type,
|
||||
*cls.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,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def expand_relationship(cls, value: Any, child: "Model") -> Any:
|
||||
return value
|
||||
27
ormar/fields/decorators.py
Normal file
27
ormar/fields/decorators.py
Normal file
@ -0,0 +1,27 @@
|
||||
from typing import Any, TYPE_CHECKING, Type
|
||||
|
||||
from ormar import ModelDefinitionError
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar.fields import BaseField
|
||||
|
||||
|
||||
class RequiredParams:
|
||||
def __init__(self, *args: str) -> None:
|
||||
self._required = list(args)
|
||||
|
||||
def __call__(self, model_field_class: Type["BaseField"]) -> Type["BaseField"]:
|
||||
old_init = model_field_class.__init__
|
||||
model_field_class._old_init = old_init
|
||||
|
||||
def __init__(instance: "BaseField", **kwargs: Any) -> None:
|
||||
super(instance.__class__, instance).__init__(**kwargs)
|
||||
for arg in self._required:
|
||||
if arg not in kwargs:
|
||||
raise ModelDefinitionError(
|
||||
f"{instance.__class__.__name__} field requires parameter: {arg}"
|
||||
)
|
||||
setattr(instance, arg, kwargs.pop(arg))
|
||||
|
||||
model_field_class.__init__ = __init__
|
||||
return model_field_class
|
||||
127
ormar/fields/foreign_key.py
Normal file
127
ormar/fields/foreign_key.py
Normal file
@ -0,0 +1,127 @@
|
||||
from typing import Any, List, Optional, TYPE_CHECKING, Type, Union, Callable
|
||||
|
||||
import sqlalchemy
|
||||
from pydantic import BaseModel
|
||||
|
||||
import ormar # noqa I101
|
||||
from ormar.exceptions import RelationshipInstanceError
|
||||
from ormar.fields.base import BaseField
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar.models import Model
|
||||
|
||||
|
||||
def create_dummy_instance(fk: Type["Model"], pk: Any = None) -> "Model":
|
||||
init_dict = {
|
||||
**{fk.Meta.pkname: pk or -1,
|
||||
'__pk_only__': True},
|
||||
**{
|
||||
k: create_dummy_instance(v.to)
|
||||
for k, v in fk.Meta.model_fields.items()
|
||||
if isinstance(v, ForeignKeyField) and not v.nullable and not v.virtual
|
||||
},
|
||||
}
|
||||
return fk(**init_dict)
|
||||
|
||||
|
||||
def ForeignKey(to, *, name: str = None, unique: bool = False, nullable: bool = True,
|
||||
related_name: str = None,
|
||||
virtual: bool = False,
|
||||
) -> Type[object]:
|
||||
fk_string = to.Meta.tablename + "." + to.Meta.pkname
|
||||
to_field = to.__fields__[to.Meta.pkname]
|
||||
namespace = dict(
|
||||
to=to,
|
||||
name=name,
|
||||
nullable=nullable,
|
||||
constraints=[sqlalchemy.schema.ForeignKey(fk_string)],
|
||||
unique=unique,
|
||||
column_type=to_field.type_.column_type,
|
||||
related_name=related_name,
|
||||
virtual=virtual,
|
||||
primary_key=False,
|
||||
index=False,
|
||||
pydantic_only=False,
|
||||
default=None,
|
||||
server_default=None
|
||||
)
|
||||
|
||||
return type("ForeignKey", (ForeignKeyField, BaseField), namespace)
|
||||
|
||||
|
||||
class ForeignKeyField(BaseField):
|
||||
to: Type["Model"]
|
||||
related_name: str
|
||||
virtual: bool
|
||||
|
||||
@classmethod
|
||||
def __get_validators__(cls) -> Callable:
|
||||
yield cls.validate
|
||||
|
||||
@classmethod
|
||||
def validate(cls, v: Any) -> Any:
|
||||
return v
|
||||
|
||||
@property
|
||||
def __type__(self) -> Type[BaseModel]:
|
||||
return self.to.__pydantic_model__
|
||||
|
||||
@classmethod
|
||||
def get_column_type(cls) -> sqlalchemy.Column:
|
||||
to_column = cls.to.Meta.model_fields[cls.to.Meta.pkname]
|
||||
return to_column.column_type
|
||||
|
||||
@classmethod
|
||||
def _extract_model_from_sequence(
|
||||
cls, value: List, child: "Model"
|
||||
) -> Union["Model", List["Model"]]:
|
||||
return [cls.expand_relationship(val, child) for val in value]
|
||||
|
||||
@classmethod
|
||||
def _register_existing_model(cls, value: "Model", child: "Model") -> "Model":
|
||||
cls.register_relation(value, child)
|
||||
return value
|
||||
|
||||
@classmethod
|
||||
def _construct_model_from_dict(cls, value: dict, child: "Model") -> "Model":
|
||||
model = cls.to(**value)
|
||||
cls.register_relation(model, child)
|
||||
return model
|
||||
|
||||
@classmethod
|
||||
def _construct_model_from_pk(cls, value: Any, child: "Model") -> "Model":
|
||||
if not isinstance(value, cls.to.pk_type()):
|
||||
raise RelationshipInstanceError(
|
||||
f"Relationship error - ForeignKey {cls.to.__name__} "
|
||||
f"is of type {cls.to.pk_type()} "
|
||||
f"while {type(value)} passed as a parameter."
|
||||
)
|
||||
model = create_dummy_instance(fk=cls.to, pk=value)
|
||||
cls.register_relation(model, child)
|
||||
return model
|
||||
|
||||
@classmethod
|
||||
def register_relation(cls, model: "Model", child: "Model") -> None:
|
||||
child_model_name = cls.related_name or child.get_name()
|
||||
model.Meta._orm_relationship_manager.add_relation(
|
||||
model, child, child_model_name, virtual=cls.virtual
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def expand_relationship(
|
||||
cls, value: Any, child: "Model"
|
||||
) -> Optional[Union["Model", List["Model"]]]:
|
||||
print("expandong relatiknship", value, child)
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
constructors = {
|
||||
f"{cls.to.__name__}": cls._register_existing_model,
|
||||
"dict": cls._construct_model_from_dict,
|
||||
"list": cls._extract_model_from_sequence,
|
||||
}
|
||||
|
||||
model = constructors.get(
|
||||
value.__class__.__name__, cls._construct_model_from_pk
|
||||
)(value, child)
|
||||
return model
|
||||
373
ormar/fields/model_fields.py
Normal file
373
ormar/fields/model_fields.py
Normal file
@ -0,0 +1,373 @@
|
||||
import datetime
|
||||
import decimal
|
||||
import re
|
||||
from typing import Type, Any, Optional
|
||||
|
||||
import pydantic
|
||||
import sqlalchemy
|
||||
from pydantic import Json
|
||||
|
||||
from ormar import ModelDefinitionError
|
||||
from ormar.fields.base import BaseField # noqa I101
|
||||
|
||||
|
||||
def is_field_nullable(nullable: Optional[bool], default: Any, server_default: Any) -> bool:
|
||||
if nullable is None:
|
||||
return default is not None or server_default is not None
|
||||
return nullable
|
||||
|
||||
|
||||
def String(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
allow_blank: bool = False,
|
||||
strip_whitespace: bool = False,
|
||||
min_length: int = None,
|
||||
max_length: int = None,
|
||||
curtail_length: int = None,
|
||||
regex: str = None,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[str]:
|
||||
if max_length is None or max_length <= 0:
|
||||
raise ModelDefinitionError(f'Parameter max_length is required for field String')
|
||||
|
||||
namespace = dict(
|
||||
__type__=str,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
allow_blank=allow_blank,
|
||||
strip_whitespace=strip_whitespace,
|
||||
min_length=min_length,
|
||||
max_length=max_length,
|
||||
curtail_length=curtail_length,
|
||||
regex=regex and re.compile(regex),
|
||||
column_type=sqlalchemy.String(length=max_length),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
|
||||
return type("String", (pydantic.ConstrainedStr, BaseField), namespace)
|
||||
|
||||
|
||||
def Integer(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
autoincrement: bool = None,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
minimum: int = None,
|
||||
maximum: int = None,
|
||||
multiple_of: int = None,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[int]:
|
||||
namespace = dict(
|
||||
__type__=int,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
ge=minimum,
|
||||
le=maximum,
|
||||
multiple_of=multiple_of,
|
||||
column_type=sqlalchemy.Integer(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=autoincrement if autoincrement is not None else primary_key
|
||||
)
|
||||
return type("Integer", (pydantic.ConstrainedInt, BaseField), namespace)
|
||||
|
||||
|
||||
def Text(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
allow_blank: bool = False,
|
||||
strip_whitespace: bool = False,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[str]:
|
||||
namespace = dict(
|
||||
__type__=str,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
allow_blank=allow_blank,
|
||||
strip_whitespace=strip_whitespace,
|
||||
column_type=sqlalchemy.Text(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
|
||||
return type("Text", (pydantic.ConstrainedStr, BaseField), namespace)
|
||||
|
||||
|
||||
def Float(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
minimum: float = None,
|
||||
maximum: float = None,
|
||||
multiple_of: int = None,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[int]:
|
||||
namespace = dict(
|
||||
__type__=float,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
ge=minimum,
|
||||
le=maximum,
|
||||
multiple_of=multiple_of,
|
||||
column_type=sqlalchemy.Float(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
return type("Float", (pydantic.ConstrainedFloat, BaseField), namespace)
|
||||
|
||||
|
||||
def Boolean(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[bool]:
|
||||
namespace = dict(
|
||||
__type__=bool,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
column_type=sqlalchemy.Boolean(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
return type("Boolean", (int, BaseField), namespace)
|
||||
|
||||
|
||||
def DateTime(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[datetime.datetime]:
|
||||
namespace = dict(
|
||||
__type__=datetime.datetime,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
column_type=sqlalchemy.DateTime(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
return type("DateTime", (datetime.datetime, BaseField), namespace)
|
||||
|
||||
|
||||
def Date(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[datetime.date]:
|
||||
namespace = dict(
|
||||
__type__=datetime.date,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
column_type=sqlalchemy.Date(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
return type("Date", (datetime.date, BaseField), namespace)
|
||||
|
||||
|
||||
def Time(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[datetime.time]:
|
||||
namespace = dict(
|
||||
__type__=datetime.time,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
column_type=sqlalchemy.Time(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
return type("Time", (datetime.time, BaseField), namespace)
|
||||
|
||||
|
||||
def JSON(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[Json]:
|
||||
namespace = dict(
|
||||
__type__=pydantic.Json,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
column_type=sqlalchemy.JSON(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
|
||||
return type("JSON", (pydantic.Json, BaseField), namespace)
|
||||
|
||||
|
||||
def BigInteger(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
autoincrement: bool = None,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
minimum: int = None,
|
||||
maximum: int = None,
|
||||
multiple_of: int = None,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
) -> Type[int]:
|
||||
namespace = dict(
|
||||
__type__=int,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
ge=minimum,
|
||||
le=maximum,
|
||||
multiple_of=multiple_of,
|
||||
column_type=sqlalchemy.BigInteger(),
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=autoincrement if autoincrement is not None else primary_key
|
||||
)
|
||||
return type("BigInteger", (pydantic.ConstrainedInt, BaseField), namespace)
|
||||
|
||||
|
||||
def Decimal(
|
||||
*,
|
||||
name: str = None,
|
||||
primary_key: bool = False,
|
||||
nullable: bool = None,
|
||||
index: bool = False,
|
||||
unique: bool = False,
|
||||
minimum: float = None,
|
||||
maximum: float = None,
|
||||
multiple_of: int = None,
|
||||
precision: int = None,
|
||||
scale: int = None,
|
||||
max_digits: int = None,
|
||||
decimal_places: int = None,
|
||||
pydantic_only: bool = False,
|
||||
default: Any = None,
|
||||
server_default: Any = None
|
||||
):
|
||||
if precision is None or precision < 0 or scale is None or scale < 0:
|
||||
raise ModelDefinitionError(f'Parameters scale and precision are required for field Decimal')
|
||||
|
||||
namespace = dict(
|
||||
__type__=decimal.Decimal,
|
||||
name=name,
|
||||
primary_key=primary_key,
|
||||
nullable=is_field_nullable(nullable, default, server_default),
|
||||
index=index,
|
||||
unique=unique,
|
||||
ge=minimum,
|
||||
le=maximum,
|
||||
multiple_of=multiple_of,
|
||||
column_type=sqlalchemy.types.DECIMAL(precision=precision, scale=scale),
|
||||
precision=precision,
|
||||
scale=scale,
|
||||
max_digits=max_digits,
|
||||
decimal_places=decimal_places,
|
||||
pydantic_only=pydantic_only,
|
||||
default=default,
|
||||
server_default=server_default,
|
||||
autoincrement=False
|
||||
)
|
||||
return type("Decimal", (pydantic.ConstrainedDecimal, BaseField), namespace)
|
||||
4
ormar/models/__init__.py
Normal file
4
ormar/models/__init__.py
Normal file
@ -0,0 +1,4 @@
|
||||
from ormar.models.fakepydantic import FakePydantic
|
||||
from ormar.models.model import Model
|
||||
|
||||
__all__ = ["FakePydantic", "Model"]
|
||||
290
ormar/models/fakepydantic.py
Normal file
290
ormar/models/fakepydantic.py
Normal file
@ -0,0 +1,290 @@
|
||||
import inspect
|
||||
import json
|
||||
import uuid
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
Dict,
|
||||
List,
|
||||
Optional,
|
||||
Set,
|
||||
TYPE_CHECKING,
|
||||
Type,
|
||||
TypeVar,
|
||||
Union, AbstractSet, Mapping,
|
||||
)
|
||||
|
||||
import databases
|
||||
import pydantic
|
||||
import sqlalchemy
|
||||
from pydantic import BaseModel
|
||||
|
||||
import ormar # noqa I100
|
||||
from ormar import ForeignKey
|
||||
from ormar.fields import BaseField
|
||||
from ormar.fields.foreign_key import ForeignKeyField
|
||||
from ormar.models.metaclass import ModelMetaclass, ModelMeta
|
||||
from ormar.relations import RelationshipManager
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar.models.model import Model
|
||||
|
||||
IntStr = Union[int, str]
|
||||
DictStrAny = Dict[str, Any]
|
||||
AbstractSetIntStr = AbstractSet[IntStr]
|
||||
MappingIntStrAny = Mapping[IntStr, Any]
|
||||
|
||||
|
||||
class FakePydantic(pydantic.BaseModel, metaclass=ModelMetaclass):
|
||||
# FakePydantic inherits from list in order to be treated as
|
||||
# request.Body parameter in fastapi routes,
|
||||
# inheriting from pydantic.BaseModel causes metaclass conflicts
|
||||
__slots__ = ('_orm_id', '_orm_saved')
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
__model_fields__: Dict[str, TypeVar[BaseField]]
|
||||
__table__: sqlalchemy.Table
|
||||
__fields__: Dict[str, pydantic.fields.ModelField]
|
||||
__pydantic_model__: Type[BaseModel]
|
||||
__pkname__: str
|
||||
__tablename__: str
|
||||
__metadata__: sqlalchemy.MetaData
|
||||
__database__: databases.Database
|
||||
_orm_relationship_manager: RelationshipManager
|
||||
Meta: ModelMeta
|
||||
|
||||
# noinspection PyMissingConstructor
|
||||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||
|
||||
object.__setattr__(self, "_orm_id", uuid.uuid4().hex)
|
||||
object.__setattr__(self, "_orm_saved", False)
|
||||
|
||||
pk_only = kwargs.pop("__pk_only__", False)
|
||||
if "pk" in kwargs:
|
||||
kwargs[self.Meta.pkname] = kwargs.pop("pk")
|
||||
kwargs = {
|
||||
k: self.Meta.model_fields[k].expand_relationship(v, self)
|
||||
for k, v in kwargs.items()
|
||||
}
|
||||
|
||||
values, fields_set, validation_error = pydantic.validate_model(
|
||||
self, kwargs
|
||||
)
|
||||
if validation_error and not pk_only:
|
||||
raise validation_error
|
||||
|
||||
object.__setattr__(self, '__dict__', values)
|
||||
object.__setattr__(self, '__fields_set__', fields_set)
|
||||
|
||||
# super().__init__(**kwargs)
|
||||
# self.values = self.__pydantic_model__(**kwargs)
|
||||
|
||||
def __del__(self) -> None:
|
||||
self.Meta._orm_relationship_manager.deregister(self)
|
||||
|
||||
def __setattr__(self, name, value):
|
||||
relation_key = self.get_name(title=True) + "_" + name
|
||||
if name in self.__slots__:
|
||||
object.__setattr__(self, name, value)
|
||||
elif name == 'pk':
|
||||
object.__setattr__(self, self.Meta.pkname, value)
|
||||
elif self.Meta._orm_relationship_manager.contains(relation_key, self):
|
||||
self.Meta.model_fields[name].expand_relationship(value, self)
|
||||
else:
|
||||
super().__setattr__(name, value)
|
||||
|
||||
def __getattr__(self, item):
|
||||
relation_key = self.get_name(title=True) + "_" + item
|
||||
if self.Meta._orm_relationship_manager.contains(relation_key, self):
|
||||
return self.Meta._orm_relationship_manager.get(relation_key, self)
|
||||
|
||||
# def __setattr__(self, key: str, value: Any) -> None:
|
||||
# if key in ('_orm_id', '_orm_relationship_manager', '_orm_saved', 'objects', '__model_fields__'):
|
||||
# return setattr(self, key, value)
|
||||
# # elif key in self._extract_related_names():
|
||||
# # value = self._convert_json(key, value, op="dumps")
|
||||
# # value = self.Meta.model_fields[key].expand_relationship(value, self)
|
||||
# # relation_key = self.get_name(title=True) + "_" + key
|
||||
# # if not self.Meta._orm_relationship_manager.contains(relation_key, self):
|
||||
# # setattr(self.values, key, value)
|
||||
# else:
|
||||
# super().__setattr__(key, value)
|
||||
|
||||
# def __getattribute__(self, key: str) -> Any:
|
||||
# if key != 'Meta' and key in self.Meta.model_fields:
|
||||
# relation_key = self.get_name(title=True) + "_" + key
|
||||
# if self.Meta._orm_relationship_manager.contains(relation_key, self):
|
||||
# return self.Meta._orm_relationship_manager.get(relation_key, self)
|
||||
# item = getattr(self.__fields__, key, None)
|
||||
# item = self._convert_json(key, item, op="loads")
|
||||
# return item
|
||||
# return super().__getattribute__(key)
|
||||
|
||||
def __same__(self, other: "Model") -> bool:
|
||||
if self.__class__ != other.__class__: # pragma no cover
|
||||
return False
|
||||
return (self._orm_id == other._orm_id or
|
||||
self.__dict__ == other.__dict__ or
|
||||
(self.pk == other.pk and self.pk is not None
|
||||
))
|
||||
|
||||
# def __repr__(self) -> str: # pragma no cover
|
||||
# return self.values.__repr__()
|
||||
|
||||
# @classmethod
|
||||
# def __get_validators__(cls) -> Callable: # pragma no cover
|
||||
# yield cls.__pydantic_model__.validate
|
||||
|
||||
@classmethod
|
||||
def get_name(cls, title: bool = False, lower: bool = True) -> str:
|
||||
name = cls.__name__
|
||||
if lower:
|
||||
name = name.lower()
|
||||
if title:
|
||||
name = name.title()
|
||||
return name
|
||||
|
||||
@property
|
||||
def pk(self) -> Any:
|
||||
return getattr(self, self.Meta.pkname)
|
||||
|
||||
@pk.setter
|
||||
def pk(self, value: Any) -> None:
|
||||
setattr(self, self.Meta.pkname, value)
|
||||
|
||||
@property
|
||||
def pk_column(self) -> sqlalchemy.Column:
|
||||
return self.Meta.table.primary_key.columns.values()[0]
|
||||
|
||||
@classmethod
|
||||
def pk_type(cls) -> Any:
|
||||
return cls.Meta.model_fields[cls.Meta.pkname].__type__
|
||||
|
||||
def dict(
|
||||
self,
|
||||
*,
|
||||
include: Union['AbstractSetIntStr', 'MappingIntStrAny'] = None,
|
||||
exclude: Union['AbstractSetIntStr', 'MappingIntStrAny'] = None,
|
||||
by_alias: bool = False,
|
||||
skip_defaults: bool = None,
|
||||
exclude_unset: bool = False,
|
||||
exclude_defaults: bool = False,
|
||||
exclude_none: bool = False,
|
||||
nested: bool = False
|
||||
) -> 'DictStrAny': # noqa: A003
|
||||
print('callin super', self.__class__)
|
||||
print('to exclude', self._exclude_related_names_not_required(nested))
|
||||
dict_instance = super().dict(include=include,
|
||||
exclude=self._exclude_related_names_not_required(nested),
|
||||
by_alias=by_alias,
|
||||
skip_defaults=skip_defaults,
|
||||
exclude_unset=exclude_unset,
|
||||
exclude_defaults=exclude_defaults,
|
||||
exclude_none=exclude_none)
|
||||
print('after super')
|
||||
for field in self._extract_related_names():
|
||||
print(self.__class__, field, nested)
|
||||
nested_model = getattr(self, field)
|
||||
|
||||
if self.Meta.model_fields[field].virtual and nested:
|
||||
continue
|
||||
if isinstance(nested_model, list) and not isinstance(
|
||||
nested_model, ormar.Model
|
||||
):
|
||||
print('nested list')
|
||||
dict_instance[field] = [x.dict(nested=True) for x in nested_model]
|
||||
else:
|
||||
print('instance')
|
||||
if nested_model is not None:
|
||||
dict_instance[field] = nested_model.dict(nested=True)
|
||||
|
||||
return dict_instance
|
||||
|
||||
def from_dict(self, value_dict: Dict) -> None:
|
||||
for key, value in value_dict.items():
|
||||
setattr(self, key, value)
|
||||
|
||||
def _convert_json(self, column_name: str, value: Any, op: str) -> Union[str, dict]:
|
||||
|
||||
if not self._is_conversion_to_json_needed(column_name):
|
||||
return value
|
||||
|
||||
condition = (
|
||||
isinstance(value, str) if op == "loads" else not isinstance(value, str)
|
||||
)
|
||||
operand = json.loads if op == "loads" else json.dumps
|
||||
|
||||
if condition:
|
||||
try:
|
||||
return operand(value)
|
||||
except TypeError: # pragma no cover
|
||||
pass
|
||||
return value
|
||||
|
||||
def _is_conversion_to_json_needed(self, column_name: str) -> bool:
|
||||
return self.Meta.model_fields.get(column_name).__type__ == pydantic.Json
|
||||
|
||||
def _extract_own_model_fields(self) -> Dict:
|
||||
related_names = self._extract_related_names()
|
||||
self_fields = {k: v for k, v in self.dict().items() if k not in related_names}
|
||||
return self_fields
|
||||
|
||||
@classmethod
|
||||
def _extract_related_names(cls) -> Set:
|
||||
related_names = set()
|
||||
for name, field in cls.Meta.model_fields.items():
|
||||
if inspect.isclass(field) and issubclass(
|
||||
field, ForeignKeyField
|
||||
):
|
||||
related_names.add(name)
|
||||
return related_names
|
||||
|
||||
@classmethod
|
||||
def _exclude_related_names_not_required(cls, nested:bool=False) -> Set:
|
||||
if nested:
|
||||
return cls._extract_related_names()
|
||||
related_names = set()
|
||||
for name, field in cls.Meta.model_fields.items():
|
||||
if inspect.isclass(field) and issubclass(field, ForeignKeyField) and field.nullable:
|
||||
related_names.add(name)
|
||||
return related_names
|
||||
|
||||
def _extract_model_db_fields(self) -> Dict:
|
||||
self_fields = self._extract_own_model_fields()
|
||||
self_fields = {
|
||||
k: v for k, v in self_fields.items() if k in self.Meta.table.columns
|
||||
}
|
||||
for field in self._extract_related_names():
|
||||
if getattr(self, field) is not None:
|
||||
self_fields[field] = getattr(
|
||||
getattr(self, field), self.Meta.model_fields[field].to.Meta.pkname
|
||||
)
|
||||
return self_fields
|
||||
|
||||
@classmethod
|
||||
def merge_instances_list(cls, result_rows: List["Model"]) -> List["Model"]:
|
||||
merged_rows = []
|
||||
for index, model in enumerate(result_rows):
|
||||
if index > 0 and model.pk == result_rows[index - 1].pk:
|
||||
result_rows[-1] = cls.merge_two_instances(model, merged_rows[-1])
|
||||
else:
|
||||
merged_rows.append(model)
|
||||
return merged_rows
|
||||
|
||||
@classmethod
|
||||
def merge_two_instances(cls, one: "Model", other: "Model") -> "Model":
|
||||
for field in one.Meta.model_fields.keys():
|
||||
if isinstance(getattr(one, field), list) and not isinstance(
|
||||
getattr(one, field), ormar.Model
|
||||
):
|
||||
setattr(other, field, getattr(one, field) + getattr(other, field))
|
||||
elif isinstance(getattr(one, field), ormar.Model):
|
||||
if getattr(one, field).pk == getattr(other, field).pk:
|
||||
setattr(
|
||||
other,
|
||||
field,
|
||||
cls.merge_two_instances(
|
||||
getattr(one, field), getattr(other, field)
|
||||
),
|
||||
)
|
||||
return other
|
||||
196
ormar/models/metaclass.py
Normal file
196
ormar/models/metaclass.py
Normal file
@ -0,0 +1,196 @@
|
||||
from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Type, Union
|
||||
|
||||
import databases
|
||||
import pydantic
|
||||
import sqlalchemy
|
||||
from pydantic import BaseConfig, create_model, Extra
|
||||
from pydantic.fields import ModelField, FieldInfo
|
||||
|
||||
from ormar import ForeignKey, ModelDefinitionError # noqa I100
|
||||
from ormar.fields import BaseField
|
||||
from ormar.fields.foreign_key import ForeignKeyField
|
||||
from ormar.queryset import QuerySet
|
||||
from ormar.relations import RelationshipManager
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar import Model
|
||||
|
||||
relationship_manager = RelationshipManager()
|
||||
|
||||
|
||||
class ModelMeta:
|
||||
tablename: str
|
||||
table: sqlalchemy.Table
|
||||
metadata: sqlalchemy.MetaData
|
||||
database: databases.Database
|
||||
columns: List[sqlalchemy.Column]
|
||||
pkname: str
|
||||
model_fields: Dict[str, Union[BaseField, ForeignKey]]
|
||||
_orm_relationship_manager: RelationshipManager
|
||||
|
||||
|
||||
def parse_pydantic_field_from_model_fields(object_dict: dict) -> Dict[str, Tuple]:
|
||||
pydantic_fields = {
|
||||
field_name: (
|
||||
base_field.__type__,
|
||||
... if base_field.is_required else base_field.default_value,
|
||||
)
|
||||
for field_name, base_field in object_dict.items()
|
||||
if isinstance(base_field, BaseField)
|
||||
}
|
||||
return pydantic_fields
|
||||
|
||||
|
||||
def register_relation_on_build(table_name: str, field: ForeignKey, name: str) -> None:
|
||||
child_relation_name = (
|
||||
field.to.get_name(title=True)
|
||||
+ "_"
|
||||
+ (field.related_name or (name.lower() + "s"))
|
||||
)
|
||||
reverse_name = child_relation_name
|
||||
relation_name = name.lower().title() + "_" + field.to.get_name()
|
||||
relationship_manager.add_relation_type(
|
||||
relation_name, reverse_name, field, table_name
|
||||
)
|
||||
|
||||
|
||||
def expand_reverse_relationships(model: Type["Model"]) -> None:
|
||||
for model_field in model.Meta.model_fields.values():
|
||||
if issubclass(model_field, ForeignKeyField):
|
||||
child_model_name = model_field.related_name or model.get_name() + "s"
|
||||
parent_model = model_field.to
|
||||
child = model
|
||||
if (
|
||||
child_model_name not in parent_model.__fields__
|
||||
and child.get_name() not in parent_model.__fields__
|
||||
):
|
||||
register_reverse_model_fields(parent_model, child, child_model_name)
|
||||
|
||||
|
||||
def register_reverse_model_fields(
|
||||
model: Type["Model"], child: Type["Model"], child_model_name: str
|
||||
) -> None:
|
||||
# model.__fields__[child_model_name] = ModelField(
|
||||
# name=child_model_name,
|
||||
# type_=Optional[Union[List[child], child]],
|
||||
# model_config=child.__config__,
|
||||
# class_validators=child.__validators__,
|
||||
# )
|
||||
model.Meta.model_fields[child_model_name] = ForeignKey(
|
||||
child, name=child_model_name, virtual=True
|
||||
)
|
||||
|
||||
|
||||
def sqlalchemy_columns_from_model_fields(
|
||||
name: str, object_dict: Dict, table_name: str
|
||||
) -> Tuple[Optional[str], List[sqlalchemy.Column], Dict[str, BaseField]]:
|
||||
columns = []
|
||||
pkname = None
|
||||
model_fields = {
|
||||
field_name: field
|
||||
for field_name, field in object_dict['__annotations__'].items()
|
||||
if issubclass(field, BaseField)
|
||||
}
|
||||
for field_name, field in model_fields.items():
|
||||
if field.primary_key:
|
||||
if pkname is not None:
|
||||
raise ModelDefinitionError("Only one primary key column is allowed.")
|
||||
if field.pydantic_only:
|
||||
raise ModelDefinitionError('Primary key column cannot be pydantic only')
|
||||
pkname = field_name
|
||||
if not field.pydantic_only:
|
||||
columns.append(field.get_column(field_name))
|
||||
if issubclass(field, ForeignKeyField):
|
||||
register_relation_on_build(table_name, field, name)
|
||||
|
||||
return pkname, columns, model_fields
|
||||
|
||||
|
||||
def populate_pydantic_default_values(attrs: Dict) -> Dict:
|
||||
for field, type_ in attrs["__annotations__"].items():
|
||||
if issubclass(type_, BaseField):
|
||||
if type_.name is None:
|
||||
type_.name = field
|
||||
def_value = type_.default_value()
|
||||
curr_def_value = attrs.get(field, 'NONE')
|
||||
print(field, curr_def_value, 'def val', type_.nullable)
|
||||
if curr_def_value == 'NONE' and isinstance(def_value, FieldInfo):
|
||||
attrs[field] = def_value
|
||||
elif curr_def_value == 'NONE' and type_.nullable:
|
||||
print(field, 'defsults tp none')
|
||||
attrs[field] = FieldInfo(default=None)
|
||||
return attrs
|
||||
|
||||
|
||||
def get_pydantic_base_orm_config() -> Type[BaseConfig]:
|
||||
class Config(BaseConfig):
|
||||
orm_mode = True
|
||||
arbitrary_types_allowed = True
|
||||
# extra = Extra.allow
|
||||
|
||||
return Config
|
||||
|
||||
|
||||
class ModelMetaclass(pydantic.main.ModelMetaclass):
|
||||
def __new__(mcs: type, name: str, bases: Any, attrs: dict) -> type:
|
||||
|
||||
attrs['Config'] = get_pydantic_base_orm_config()
|
||||
new_model = super().__new__( # type: ignore
|
||||
mcs, name, bases, attrs
|
||||
)
|
||||
|
||||
if hasattr(new_model, 'Meta'):
|
||||
|
||||
if attrs.get("__abstract__"):
|
||||
return new_model
|
||||
|
||||
annotations = attrs.get("__annotations__") or new_model.__annotations__
|
||||
attrs["__annotations__"]= annotations
|
||||
attrs = populate_pydantic_default_values(attrs)
|
||||
|
||||
print(attrs)
|
||||
|
||||
tablename = name.lower() + "s"
|
||||
new_model.Meta.tablename = new_model.Meta.tablename or tablename
|
||||
|
||||
# sqlalchemy table creation
|
||||
|
||||
pkname, columns, model_fields = sqlalchemy_columns_from_model_fields(
|
||||
name, attrs, new_model.Meta.tablename
|
||||
)
|
||||
|
||||
if hasattr(new_model.Meta, "model_fields") and not pkname:
|
||||
model_fields = new_model.Meta.model_fields
|
||||
for fieldname, field in new_model.Meta.model_fields.items():
|
||||
if field.primary_key:
|
||||
pkname=fieldname
|
||||
columns = new_model.Meta.table.columns
|
||||
|
||||
if not hasattr(new_model.Meta, "table"):
|
||||
new_model.Meta.table = sqlalchemy.Table(new_model.Meta.tablename, new_model.Meta.metadata, *columns)
|
||||
|
||||
new_model.Meta.columns = columns
|
||||
new_model.Meta.pkname = pkname
|
||||
|
||||
if not pkname:
|
||||
raise ModelDefinitionError("Table has to have a primary key.")
|
||||
|
||||
# pydantic model creation
|
||||
new_model.Meta.pydantic_fields = parse_pydantic_field_from_model_fields(attrs)
|
||||
new_model.Meta.pydantic_model = create_model(
|
||||
name, __config__=get_pydantic_base_orm_config(), **new_model.Meta.pydantic_fields
|
||||
)
|
||||
|
||||
new_model.Meta.model_fields = model_fields
|
||||
print(attrs, 'before super')
|
||||
print(new_model.Meta.__dict__)
|
||||
new_model = super().__new__( # type: ignore
|
||||
mcs, name, bases, attrs
|
||||
)
|
||||
expand_reverse_relationships(new_model)
|
||||
|
||||
new_model.Meta._orm_relationship_manager = relationship_manager
|
||||
new_model.objects = QuerySet(new_model)
|
||||
|
||||
# breakpoint()
|
||||
return new_model
|
||||
85
ormar/models/model.py
Normal file
85
ormar/models/model.py
Normal file
@ -0,0 +1,85 @@
|
||||
from typing import Any, List
|
||||
|
||||
import sqlalchemy
|
||||
|
||||
import ormar.queryset # noqa I100
|
||||
from ormar.models import FakePydantic # noqa I100
|
||||
|
||||
|
||||
class Model(FakePydantic):
|
||||
__abstract__ = False
|
||||
|
||||
# objects = ormar.queryset.QuerySet()
|
||||
|
||||
@classmethod
|
||||
def from_row(
|
||||
cls,
|
||||
row: sqlalchemy.engine.ResultProxy,
|
||||
select_related: List = None,
|
||||
previous_table: str = None,
|
||||
) -> "Model":
|
||||
|
||||
item = {}
|
||||
select_related = select_related or []
|
||||
|
||||
table_prefix = cls.Meta._orm_relationship_manager.resolve_relation_join(
|
||||
previous_table, cls.Meta.table.name
|
||||
)
|
||||
previous_table = cls.Meta.table.name
|
||||
for related in select_related:
|
||||
if "__" in related:
|
||||
first_part, remainder = related.split("__", 1)
|
||||
model_cls = cls.Meta.model_fields[first_part].to
|
||||
child = model_cls.from_row(
|
||||
row, select_related=[remainder], previous_table=previous_table
|
||||
)
|
||||
item[first_part] = child
|
||||
else:
|
||||
model_cls = cls.Meta.model_fields[related].to
|
||||
child = model_cls.from_row(row, previous_table=previous_table)
|
||||
item[related] = child
|
||||
|
||||
for column in cls.Meta.table.columns:
|
||||
if column.name not in item:
|
||||
item[column.name] = row[
|
||||
f'{table_prefix + "_" if table_prefix else ""}{column.name}'
|
||||
]
|
||||
|
||||
return cls(**item)
|
||||
|
||||
async def save(self) -> "Model":
|
||||
self_fields = self._extract_model_db_fields()
|
||||
if self.Meta.model_fields.get(self.Meta.pkname).autoincrement:
|
||||
self_fields.pop(self.Meta.pkname, None)
|
||||
expr = self.Meta.table.insert()
|
||||
expr = expr.values(**self_fields)
|
||||
item_id = await self.Meta.database.execute(expr)
|
||||
setattr(self, self.Meta.pkname, item_id)
|
||||
return self
|
||||
|
||||
async def update(self, **kwargs: Any) -> int:
|
||||
if kwargs:
|
||||
new_values = {**self.dict(), **kwargs}
|
||||
self.from_dict(new_values)
|
||||
|
||||
self_fields = self._extract_model_db_fields()
|
||||
self_fields.pop(self.Meta.pkname)
|
||||
expr = (
|
||||
self.Meta.table.update()
|
||||
.values(**self_fields)
|
||||
.where(self.pk_column == getattr(self, self.Meta.pkname))
|
||||
)
|
||||
result = await self.Meta.database.execute(expr)
|
||||
return result
|
||||
|
||||
async def delete(self) -> int:
|
||||
expr = self.Meta.table.delete()
|
||||
expr = expr.where(self.pk_column == (getattr(self, self.Meta.pkname)))
|
||||
result = await self.Meta.database.execute(expr)
|
||||
return result
|
||||
|
||||
async def load(self) -> "Model":
|
||||
expr = self.Meta.table.select().where(self.pk_column == self.pk)
|
||||
row = await self.Meta.database.fetch_one(expr)
|
||||
self.from_dict(dict(row))
|
||||
return self
|
||||
3
ormar/queryset/__init__.py
Normal file
3
ormar/queryset/__init__.py
Normal file
@ -0,0 +1,3 @@
|
||||
from ormar.queryset.queryset import QuerySet
|
||||
|
||||
__all__ = ["QuerySet"]
|
||||
178
ormar/queryset/clause.py
Normal file
178
ormar/queryset/clause.py
Normal file
@ -0,0 +1,178 @@
|
||||
from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Type, Union
|
||||
|
||||
import sqlalchemy
|
||||
from sqlalchemy import text
|
||||
|
||||
import ormar # noqa I100
|
||||
from ormar.exceptions import QueryDefinitionError
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar import Model
|
||||
|
||||
FILTER_OPERATORS = {
|
||||
"exact": "__eq__",
|
||||
"iexact": "ilike",
|
||||
"contains": "like",
|
||||
"icontains": "ilike",
|
||||
"in": "in_",
|
||||
"gt": "__gt__",
|
||||
"gte": "__ge__",
|
||||
"lt": "__lt__",
|
||||
"lte": "__le__",
|
||||
}
|
||||
ESCAPE_CHARACTERS = ["%", "_"]
|
||||
|
||||
|
||||
class QueryClause:
|
||||
def __init__(
|
||||
self, model_cls: Type["Model"], filter_clauses: List, select_related: List,
|
||||
) -> None:
|
||||
|
||||
self._select_related = select_related
|
||||
self.filter_clauses = filter_clauses
|
||||
|
||||
self.model_cls = model_cls
|
||||
self.table = self.model_cls.Meta.table
|
||||
|
||||
def filter( # noqa: A003
|
||||
self, **kwargs: Any
|
||||
) -> Tuple[List[sqlalchemy.sql.expression.TextClause], List[str]]:
|
||||
filter_clauses = self.filter_clauses
|
||||
select_related = list(self._select_related)
|
||||
|
||||
if kwargs.get("pk"):
|
||||
pk_name = self.model_cls.Meta.pkname
|
||||
kwargs[pk_name] = kwargs.pop("pk")
|
||||
|
||||
for key, value in kwargs.items():
|
||||
table_prefix = ""
|
||||
if "__" in key:
|
||||
parts = key.split("__")
|
||||
|
||||
(
|
||||
op,
|
||||
field_name,
|
||||
related_parts,
|
||||
) = self._extract_operator_field_and_related(parts)
|
||||
|
||||
model_cls = self.model_cls
|
||||
if related_parts:
|
||||
(
|
||||
select_related,
|
||||
table_prefix,
|
||||
model_cls,
|
||||
) = self._determine_filter_target_table(
|
||||
related_parts, select_related
|
||||
)
|
||||
|
||||
table = model_cls.Meta.table
|
||||
column = model_cls.Meta.table.columns[field_name]
|
||||
|
||||
else:
|
||||
op = "exact"
|
||||
column = self.table.columns[key]
|
||||
table = self.table
|
||||
|
||||
value, has_escaped_character = self._escape_characters_in_clause(op, value)
|
||||
|
||||
if isinstance(value, ormar.Model):
|
||||
value = value.pk
|
||||
|
||||
op_attr = FILTER_OPERATORS[op]
|
||||
clause = getattr(column, op_attr)(value)
|
||||
clause = self._compile_clause(
|
||||
clause,
|
||||
column,
|
||||
table,
|
||||
table_prefix,
|
||||
modifiers={"escape": "\\" if has_escaped_character else None},
|
||||
)
|
||||
filter_clauses.append(clause)
|
||||
|
||||
return filter_clauses, select_related
|
||||
|
||||
def _determine_filter_target_table(
|
||||
self, related_parts: List[str], select_related: List[str]
|
||||
) -> Tuple[List[str], str, "Model"]:
|
||||
|
||||
table_prefix = ""
|
||||
model_cls = self.model_cls
|
||||
select_related = [relation for relation in select_related]
|
||||
|
||||
# Add any implied select_related
|
||||
related_str = "__".join(related_parts)
|
||||
if related_str not in select_related:
|
||||
select_related.append(related_str)
|
||||
|
||||
# Walk the relationships to the actual model class
|
||||
# against which the comparison is being made.
|
||||
previous_table = model_cls.Meta.tablename
|
||||
for part in related_parts:
|
||||
current_table = model_cls.Meta.model_fields[part].to.Meta.tablename
|
||||
manager = model_cls.Meta._orm_relationship_manager
|
||||
table_prefix = manager.resolve_relation_join(previous_table, current_table)
|
||||
model_cls = model_cls.Meta.model_fields[part].to
|
||||
previous_table = current_table
|
||||
return select_related, table_prefix, model_cls
|
||||
|
||||
def _compile_clause(
|
||||
self,
|
||||
clause: sqlalchemy.sql.expression.BinaryExpression,
|
||||
column: sqlalchemy.Column,
|
||||
table: sqlalchemy.Table,
|
||||
table_prefix: str,
|
||||
modifiers: Dict,
|
||||
) -> sqlalchemy.sql.expression.TextClause:
|
||||
for modifier, modifier_value in modifiers.items():
|
||||
clause.modifiers[modifier] = modifier_value
|
||||
|
||||
clause_text = str(
|
||||
clause.compile(
|
||||
dialect=self.model_cls.Meta.database._backend._dialect,
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
alias = f"{table_prefix}_" if table_prefix else ""
|
||||
aliased_name = f"{alias}{table.name}.{column.name}"
|
||||
clause_text = clause_text.replace(f"{table.name}.{column.name}", aliased_name)
|
||||
clause = text(clause_text)
|
||||
return clause
|
||||
|
||||
@staticmethod
|
||||
def _escape_characters_in_clause(
|
||||
op: str, value: Union[str, "Model"]
|
||||
) -> Tuple[str, bool]:
|
||||
has_escaped_character = False
|
||||
|
||||
if op not in ["contains", "icontains"]:
|
||||
return value, has_escaped_character
|
||||
|
||||
if isinstance(value, ormar.Model):
|
||||
raise QueryDefinitionError(
|
||||
"You cannot use contains and icontains with instance of the Model"
|
||||
)
|
||||
|
||||
has_escaped_character = any(c for c in ESCAPE_CHARACTERS if c in value)
|
||||
|
||||
if has_escaped_character:
|
||||
# enable escape modifier
|
||||
for char in ESCAPE_CHARACTERS:
|
||||
value = value.replace(char, f"\\{char}")
|
||||
value = f"%{value}%"
|
||||
|
||||
return value, has_escaped_character
|
||||
|
||||
@staticmethod
|
||||
def _extract_operator_field_and_related(
|
||||
parts: List[str],
|
||||
) -> Tuple[str, str, Optional[List]]:
|
||||
if parts[-1] in FILTER_OPERATORS:
|
||||
op = parts[-1]
|
||||
field_name = parts[-2]
|
||||
related_parts = parts[:-2]
|
||||
else:
|
||||
op = "exact"
|
||||
field_name = parts[-1]
|
||||
related_parts = parts[:-1]
|
||||
|
||||
return op, field_name, related_parts
|
||||
236
ormar/queryset/query.py
Normal file
236
ormar/queryset/query.py
Normal file
@ -0,0 +1,236 @@
|
||||
from typing import List, NamedTuple, TYPE_CHECKING, Tuple, Type
|
||||
|
||||
import sqlalchemy
|
||||
from sqlalchemy import text
|
||||
|
||||
import ormar # noqa I100
|
||||
from ormar.fields import BaseField
|
||||
from ormar.fields.foreign_key import ForeignKeyField
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar import Model
|
||||
|
||||
|
||||
class JoinParameters(NamedTuple):
|
||||
prev_model: Type["Model"]
|
||||
previous_alias: str
|
||||
from_table: str
|
||||
model_cls: Type["Model"]
|
||||
|
||||
|
||||
class Query:
|
||||
def __init__(
|
||||
self,
|
||||
model_cls: Type["Model"],
|
||||
filter_clauses: List,
|
||||
select_related: List,
|
||||
limit_count: int,
|
||||
offset: int,
|
||||
) -> None:
|
||||
|
||||
self.query_offset = offset
|
||||
self.limit_count = limit_count
|
||||
self._select_related = select_related
|
||||
self.filter_clauses = filter_clauses
|
||||
|
||||
self.model_cls = model_cls
|
||||
self.table = self.model_cls.Meta.table
|
||||
|
||||
self.auto_related = []
|
||||
self.used_aliases = []
|
||||
self.already_checked = []
|
||||
|
||||
self.select_from = None
|
||||
self.columns = None
|
||||
self.order_bys = None
|
||||
|
||||
def build_select_expression(self) -> Tuple[sqlalchemy.sql.select, List[str]]:
|
||||
self.columns = list(self.table.columns)
|
||||
self.order_bys = [text(f"{self.table.name}.{self.model_cls.Meta.pkname}")]
|
||||
self.select_from = self.table
|
||||
|
||||
for key in self.model_cls.Meta.model_fields:
|
||||
if (
|
||||
not self.model_cls.Meta.model_fields[key].nullable
|
||||
and isinstance(
|
||||
self.model_cls.Meta.model_fields[key], ForeignKeyField,
|
||||
)
|
||||
and key not in self._select_related
|
||||
):
|
||||
self._select_related = [key] + self._select_related
|
||||
|
||||
start_params = JoinParameters(
|
||||
self.model_cls, "", self.table.name, self.model_cls
|
||||
)
|
||||
self._extract_auto_required_relations(prev_model=start_params.prev_model)
|
||||
self._include_auto_related_models()
|
||||
self._select_related.sort(key=lambda item: (-len(item), item))
|
||||
|
||||
for item in self._select_related:
|
||||
join_parameters = JoinParameters(
|
||||
self.model_cls, "", self.table.name, self.model_cls
|
||||
)
|
||||
|
||||
for part in item.split("__"):
|
||||
join_parameters = self._build_join_parameters(part, join_parameters)
|
||||
|
||||
expr = sqlalchemy.sql.select(self.columns)
|
||||
expr = expr.select_from(self.select_from)
|
||||
|
||||
expr = self._apply_expression_modifiers(expr)
|
||||
|
||||
print(expr.compile(compile_kwargs={"literal_binds": True}))
|
||||
self._reset_query_parameters()
|
||||
|
||||
return expr, self._select_related
|
||||
|
||||
@staticmethod
|
||||
def prefixed_columns(alias: str, table: sqlalchemy.Table) -> List[text]:
|
||||
return [
|
||||
text(f"{alias}_{table.name}.{column.name} as {alias}_{column.name}")
|
||||
for column in table.columns
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def prefixed_table_name(alias: str, name: str) -> text:
|
||||
return text(f"{name} {alias}_{name}")
|
||||
|
||||
@staticmethod
|
||||
def _field_is_a_foreign_key_and_no_circular_reference(
|
||||
field: BaseField, field_name: str, rel_part: str
|
||||
) -> bool:
|
||||
return issubclass(field, ForeignKeyField) and field_name not in rel_part
|
||||
|
||||
def _field_qualifies_to_deeper_search(
|
||||
self, field: ForeignKeyField, parent_virtual: bool, nested: bool, rel_part: str
|
||||
) -> bool:
|
||||
prev_part_of_related = "__".join(rel_part.split("__")[:-1])
|
||||
partial_match = any(
|
||||
[x.startswith(prev_part_of_related) for x in self._select_related]
|
||||
)
|
||||
already_checked = any(
|
||||
[x.startswith(rel_part) for x in (self.auto_related + self.already_checked)]
|
||||
)
|
||||
return (
|
||||
(field.virtual and parent_virtual)
|
||||
or (partial_match and not already_checked)
|
||||
) or not nested
|
||||
|
||||
def on_clause(
|
||||
self, previous_alias: str, alias: str, from_clause: str, to_clause: str,
|
||||
) -> text:
|
||||
left_part = f"{alias}_{to_clause}"
|
||||
right_part = f"{previous_alias + '_' if previous_alias else ''}{from_clause}"
|
||||
return text(f"{left_part}={right_part}")
|
||||
|
||||
def _build_join_parameters(
|
||||
self, part: str, join_params: JoinParameters
|
||||
) -> JoinParameters:
|
||||
model_cls = join_params.model_cls.Meta.model_fields[part].to
|
||||
to_table = model_cls.Meta.table.name
|
||||
|
||||
alias = model_cls.Meta._orm_relationship_manager.resolve_relation_join(
|
||||
join_params.from_table, to_table
|
||||
)
|
||||
if alias not in self.used_aliases:
|
||||
if join_params.prev_model.Meta.model_fields[part].virtual:
|
||||
to_key = next(
|
||||
(
|
||||
v
|
||||
for k, v in model_cls.Meta.model_fields.items()
|
||||
if issubclass(v, ForeignKeyField) and v.to == join_params.prev_model
|
||||
),
|
||||
None,
|
||||
).name
|
||||
from_key = model_cls.Meta.pkname
|
||||
else:
|
||||
to_key = model_cls.Meta.pkname
|
||||
from_key = part
|
||||
|
||||
on_clause = self.on_clause(
|
||||
previous_alias=join_params.previous_alias,
|
||||
alias=alias,
|
||||
from_clause=f"{join_params.from_table}.{from_key}",
|
||||
to_clause=f"{to_table}.{to_key}",
|
||||
)
|
||||
target_table = self.prefixed_table_name(alias, to_table)
|
||||
self.select_from = sqlalchemy.sql.outerjoin(
|
||||
self.select_from, target_table, on_clause
|
||||
)
|
||||
self.order_bys.append(text(f"{alias}_{to_table}.{model_cls.Meta.pkname}"))
|
||||
self.columns.extend(self.prefixed_columns(alias, model_cls.Meta.table))
|
||||
self.used_aliases.append(alias)
|
||||
|
||||
previous_alias = alias
|
||||
from_table = to_table
|
||||
prev_model = model_cls
|
||||
return JoinParameters(prev_model, previous_alias, from_table, model_cls)
|
||||
|
||||
def _extract_auto_required_relations(
|
||||
self,
|
||||
prev_model: Type["Model"],
|
||||
rel_part: str = "",
|
||||
nested: bool = False,
|
||||
parent_virtual: bool = False,
|
||||
) -> None:
|
||||
for field_name, field in prev_model.Meta.model_fields.items():
|
||||
if self._field_is_a_foreign_key_and_no_circular_reference(
|
||||
field, field_name, rel_part
|
||||
):
|
||||
rel_part = field_name if not rel_part else rel_part + "__" + field_name
|
||||
if not field.nullable:
|
||||
print('add', rel_part, field)
|
||||
if rel_part not in self._select_related:
|
||||
new_related = "__".join(rel_part.split("__")[:-1]) if len(
|
||||
rel_part.split("__")) > 1 else rel_part
|
||||
self.auto_related.append(new_related)
|
||||
rel_part = ""
|
||||
elif self._field_qualifies_to_deeper_search(
|
||||
field, parent_virtual, nested, rel_part
|
||||
):
|
||||
print('deeper', rel_part, field, field.to)
|
||||
self._extract_auto_required_relations(
|
||||
prev_model=field.to,
|
||||
rel_part=rel_part,
|
||||
nested=True,
|
||||
parent_virtual=field.virtual,
|
||||
)
|
||||
else:
|
||||
self.already_checked.append(rel_part)
|
||||
rel_part = ""
|
||||
|
||||
def _include_auto_related_models(self) -> None:
|
||||
if self.auto_related:
|
||||
new_joins = []
|
||||
for join in self._select_related:
|
||||
if not any([x.startswith(join) for x in self.auto_related]):
|
||||
new_joins.append(join)
|
||||
self._select_related = new_joins + self.auto_related
|
||||
|
||||
def _apply_expression_modifiers(
|
||||
self, expr: sqlalchemy.sql.select
|
||||
) -> sqlalchemy.sql.select:
|
||||
if self.filter_clauses:
|
||||
if len(self.filter_clauses) == 1:
|
||||
clause = self.filter_clauses[0]
|
||||
else:
|
||||
clause = sqlalchemy.sql.and_(*self.filter_clauses)
|
||||
expr = expr.where(clause)
|
||||
|
||||
if self.limit_count:
|
||||
expr = expr.limit(self.limit_count)
|
||||
|
||||
if self.query_offset:
|
||||
expr = expr.offset(self.query_offset)
|
||||
|
||||
for order in self.order_bys:
|
||||
expr = expr.order_by(order)
|
||||
return expr
|
||||
|
||||
def _reset_query_parameters(self) -> None:
|
||||
self.select_from = None
|
||||
self.columns = None
|
||||
self.order_bys = None
|
||||
self.auto_related = []
|
||||
self.used_aliases = []
|
||||
self.already_checked = []
|
||||
181
ormar/queryset/queryset.py
Normal file
181
ormar/queryset/queryset.py
Normal file
@ -0,0 +1,181 @@
|
||||
from typing import Any, List, TYPE_CHECKING, Tuple, Type, Union
|
||||
|
||||
import databases
|
||||
import sqlalchemy
|
||||
|
||||
import ormar # noqa I100
|
||||
from ormar import MultipleMatches, NoMatch
|
||||
from ormar.queryset.clause import QueryClause
|
||||
from ormar.queryset.query import Query
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar import Model
|
||||
|
||||
|
||||
class QuerySet:
|
||||
def __init__(
|
||||
self,
|
||||
model_cls: Type["Model"] = None,
|
||||
filter_clauses: List = None,
|
||||
select_related: List = None,
|
||||
limit_count: int = None,
|
||||
offset: int = None,
|
||||
) -> None:
|
||||
self.model_cls = model_cls
|
||||
self.filter_clauses = [] if filter_clauses is None else filter_clauses
|
||||
self._select_related = [] if select_related is None else select_related
|
||||
self.limit_count = limit_count
|
||||
self.query_offset = offset
|
||||
self.order_bys = None
|
||||
|
||||
def __get__(self, instance: "QuerySet", owner: Type["Model"]) -> "QuerySet":
|
||||
return self.__class__(model_cls=owner)
|
||||
|
||||
@property
|
||||
def database(self) -> databases.Database:
|
||||
return self.model_cls.Meta.database
|
||||
|
||||
@property
|
||||
def table(self) -> sqlalchemy.Table:
|
||||
return self.model_cls.Meta.table
|
||||
|
||||
def build_select_expression(self) -> sqlalchemy.sql.select:
|
||||
qry = Query(
|
||||
model_cls=self.model_cls,
|
||||
select_related=self._select_related,
|
||||
filter_clauses=self.filter_clauses,
|
||||
offset=self.query_offset,
|
||||
limit_count=self.limit_count,
|
||||
)
|
||||
exp, self._select_related = qry.build_select_expression()
|
||||
return exp
|
||||
|
||||
def filter(self, **kwargs: Any) -> "QuerySet": # noqa: A003
|
||||
qryclause = QueryClause(
|
||||
model_cls=self.model_cls,
|
||||
select_related=self._select_related,
|
||||
filter_clauses=self.filter_clauses,
|
||||
)
|
||||
filter_clauses, select_related = qryclause.filter(**kwargs)
|
||||
|
||||
return self.__class__(
|
||||
model_cls=self.model_cls,
|
||||
filter_clauses=filter_clauses,
|
||||
select_related=select_related,
|
||||
limit_count=self.limit_count,
|
||||
offset=self.query_offset,
|
||||
)
|
||||
|
||||
def select_related(self, related: Union[List, Tuple, str]) -> "QuerySet":
|
||||
if not isinstance(related, (list, tuple)):
|
||||
related = [related]
|
||||
|
||||
related = list(self._select_related) + related
|
||||
return self.__class__(
|
||||
model_cls=self.model_cls,
|
||||
filter_clauses=self.filter_clauses,
|
||||
select_related=related,
|
||||
limit_count=self.limit_count,
|
||||
offset=self.query_offset,
|
||||
)
|
||||
|
||||
async def exists(self) -> bool:
|
||||
expr = self.build_select_expression()
|
||||
expr = sqlalchemy.exists(expr).select()
|
||||
return await self.database.fetch_val(expr)
|
||||
|
||||
async def count(self) -> int:
|
||||
expr = self.build_select_expression().alias("subquery_for_count")
|
||||
expr = sqlalchemy.func.count().select().select_from(expr)
|
||||
return await self.database.fetch_val(expr)
|
||||
|
||||
def limit(self, limit_count: int) -> "QuerySet":
|
||||
return self.__class__(
|
||||
model_cls=self.model_cls,
|
||||
filter_clauses=self.filter_clauses,
|
||||
select_related=self._select_related,
|
||||
limit_count=limit_count,
|
||||
offset=self.query_offset,
|
||||
)
|
||||
|
||||
def offset(self, offset: int) -> "QuerySet":
|
||||
return self.__class__(
|
||||
model_cls=self.model_cls,
|
||||
filter_clauses=self.filter_clauses,
|
||||
select_related=self._select_related,
|
||||
limit_count=self.limit_count,
|
||||
offset=offset,
|
||||
)
|
||||
|
||||
async def first(self, **kwargs: Any) -> "Model":
|
||||
if kwargs:
|
||||
return await self.filter(**kwargs).first()
|
||||
|
||||
rows = await self.limit(1).all()
|
||||
if rows:
|
||||
return rows[0]
|
||||
|
||||
async def get(self, **kwargs: Any) -> "Model":
|
||||
if kwargs:
|
||||
return await self.filter(**kwargs).get()
|
||||
|
||||
expr = self.build_select_expression().limit(2)
|
||||
rows = await self.database.fetch_all(expr)
|
||||
|
||||
if not rows:
|
||||
raise NoMatch()
|
||||
if len(rows) > 1:
|
||||
raise MultipleMatches()
|
||||
return self.model_cls.from_row(rows[0], select_related=self._select_related)
|
||||
|
||||
async def all(self, **kwargs: Any) -> List["Model"]: # noqa: A003
|
||||
if kwargs:
|
||||
return await self.filter(**kwargs).all()
|
||||
|
||||
expr = self.build_select_expression()
|
||||
rows = await self.database.fetch_all(expr)
|
||||
result_rows = [
|
||||
self.model_cls.from_row(row, select_related=self._select_related)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
result_rows = self.model_cls.merge_instances_list(result_rows)
|
||||
|
||||
return result_rows
|
||||
|
||||
async def create(self, **kwargs: Any) -> "Model":
|
||||
|
||||
new_kwargs = dict(**kwargs)
|
||||
|
||||
# Remove primary key when None to prevent not null constraint in postgresql.
|
||||
pkname = self.model_cls.Meta.pkname
|
||||
pk = self.model_cls.Meta.model_fields[pkname]
|
||||
if (
|
||||
pkname in new_kwargs
|
||||
and new_kwargs.get(pkname) is None
|
||||
and (pk.nullable or pk.autoincrement)
|
||||
):
|
||||
del new_kwargs[pkname]
|
||||
|
||||
# substitute related models with their pk
|
||||
for field in self.model_cls._extract_related_names():
|
||||
if field in new_kwargs and new_kwargs.get(field) is not None:
|
||||
if isinstance(new_kwargs.get(field), ormar.Model):
|
||||
new_kwargs[field] = getattr(
|
||||
new_kwargs.get(field),
|
||||
self.model_cls.Meta.model_fields[field].to.Meta.pkname,
|
||||
)
|
||||
else:
|
||||
new_kwargs[field] = new_kwargs.get(field).get(
|
||||
self.model_cls.Meta.model_fields[field].to.Meta.pkname
|
||||
)
|
||||
|
||||
# Build the insert expression.
|
||||
expr = self.table.insert()
|
||||
expr = expr.values(**new_kwargs)
|
||||
|
||||
# Execute the insert, and return a new model instance.
|
||||
instance = self.model_cls(**kwargs)
|
||||
pk = await self.database.execute(expr)
|
||||
setattr(instance, self.model_cls.Meta.pkname, pk)
|
||||
return instance
|
||||
104
ormar/relations.py
Normal file
104
ormar/relations.py
Normal file
@ -0,0 +1,104 @@
|
||||
import pprint
|
||||
import string
|
||||
import uuid
|
||||
from random import choices
|
||||
from typing import List, TYPE_CHECKING, Union
|
||||
from weakref import proxy
|
||||
|
||||
from ormar import ForeignKey
|
||||
from ormar.fields.foreign_key import ForeignKeyField
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
from ormar.models import FakePydantic, Model
|
||||
|
||||
|
||||
def get_table_alias() -> str:
|
||||
return "".join(choices(string.ascii_uppercase, k=2)) + uuid.uuid4().hex[:4]
|
||||
|
||||
|
||||
class RelationshipManager:
|
||||
def __init__(self) -> None:
|
||||
self._relations = dict()
|
||||
self._aliases = dict()
|
||||
|
||||
def add_relation_type(
|
||||
self, relations_key: str, reverse_key: str, field: ForeignKeyField, table_name: str
|
||||
) -> None:
|
||||
if relations_key not in self._relations:
|
||||
self._relations[relations_key] = {"type": "primary"}
|
||||
self._aliases[f"{table_name}_{field.to.Meta.tablename}"] = get_table_alias()
|
||||
if reverse_key not in self._relations:
|
||||
self._relations[reverse_key] = {"type": "reverse"}
|
||||
self._aliases[f"{field.to.Meta.tablename}_{table_name}"] = get_table_alias()
|
||||
|
||||
def deregister(self, model: "FakePydantic") -> None:
|
||||
for rel_type in self._relations.keys():
|
||||
if model.get_name() in rel_type.lower():
|
||||
if model._orm_id in self._relations[rel_type]:
|
||||
del self._relations[rel_type][model._orm_id]
|
||||
|
||||
def add_relation(
|
||||
self,
|
||||
parent: "FakePydantic",
|
||||
child: "FakePydantic",
|
||||
child_model_name: str,
|
||||
virtual: bool = False,
|
||||
) -> None:
|
||||
parent_id, child_id = parent._orm_id, child._orm_id
|
||||
parent_name = parent.get_name(title=True)
|
||||
child_name = (
|
||||
child_model_name
|
||||
if child.get_name() != child_model_name
|
||||
else child.get_name() + "s"
|
||||
)
|
||||
if virtual:
|
||||
child_name, parent_name = parent_name, child.get_name()
|
||||
child_id, parent_id = parent_id, child_id
|
||||
child, parent = parent, proxy(child)
|
||||
child_name = child_name.lower() + "s"
|
||||
else:
|
||||
child = proxy(child)
|
||||
|
||||
parent_relation_name = parent_name.title() + "_" + child_name
|
||||
parents_list = self._relations[parent_relation_name].setdefault(parent_id, [])
|
||||
self.append_related_model(parents_list, child)
|
||||
|
||||
child_relation_name = child.get_name(title=True) + "_" + parent_name.lower()
|
||||
children_list = self._relations[child_relation_name].setdefault(child_id, [])
|
||||
self.append_related_model(children_list, parent)
|
||||
|
||||
@staticmethod
|
||||
def append_related_model(relations_list: List["Model"], model: "Model") -> None:
|
||||
print("appending", relations_list, model)
|
||||
for relation_child in relations_list:
|
||||
try:
|
||||
print(relation_child.__same__(model), "same")
|
||||
if relation_child.__same__(model):
|
||||
return
|
||||
except ReferenceError:
|
||||
continue
|
||||
|
||||
relations_list.append(model)
|
||||
|
||||
def contains(self, relations_key: str, instance: "FakePydantic") -> bool:
|
||||
if relations_key in self._relations:
|
||||
return instance._orm_id in self._relations[relations_key]
|
||||
return False
|
||||
|
||||
def get(
|
||||
self, relations_key: str, instance: "FakePydantic"
|
||||
) -> Union["Model", List["Model"]]:
|
||||
if relations_key in self._relations:
|
||||
if instance._orm_id in self._relations[relations_key]:
|
||||
if self._relations[relations_key]["type"] == "primary":
|
||||
return self._relations[relations_key][instance._orm_id][0]
|
||||
return self._relations[relations_key][instance._orm_id]
|
||||
|
||||
def resolve_relation_join(self, from_table: str, to_table: str) -> str:
|
||||
return self._aliases.get(f"{from_table}_{to_table}", "")
|
||||
|
||||
def __str__(self) -> str: # pragma no cover
|
||||
return pprint.pformat(self._relations, indent=4, width=1)
|
||||
|
||||
def __repr__(self) -> str: # pragma no cover
|
||||
return self.__str__()
|
||||
Reference in New Issue
Block a user