Я пытаюсь создать абстрактный класс репозитория, чтобы резко сократить длину своего стандартного кода. Между тем, для этого мне нужно связать границы Entity с определенным Domain (например, мы используем Pick в Typescript), чтобы привлечь мое внимание к таким Missing positional arguments. Возможно ли это сделать в Python?
"""
Generic Repository
"""
from src.Managers.session import SessionManager
from typing import Generic, TypeVar, Union
from sqlalchemy.orm import Query
from src.User.models.TermsOfUse import TermsOfUseModel, TermsOfUse
from src.User.models.User import UserModel, User
entity_bound = Union[UserModel, TermsOfUseModel]
domain_bound = Union[User, TermsOfUse]
Entity = TypeVar('Entity', bound=entity_bound)
Domain = TypeVar('Domain', bound=domain_bound)
SpecificException = TypeVar('SpecificException', bound=Exception)
class RepoPattern(Generic[Entity, Domain, SpecificException]):
"""
Repository Pattern.
"""
def __init__(
self,
entity: type[Entity],
domain: type[Domain],
exception: type[SpecificException],
session_context: SessionManager
):
self._entity = entity
self._domain = domain
self._exception = exception
self._session_context = session_context
@property
def session_context(self) -> SessionManager:
return self._session_context
def add(self, domain: Domain) -> None:
"""
Add a new domain object.
Parameters:
----
:param domain: domain object.
Returns:
----
Domain
"""
entity = self._get_entity_from_domain(domain)
self._session_context.add(entity)
def modify(self, domain: Domain) -> None:
"""
Add a new domain object.
Parameters:
----
:param domain: domain object.
Returns:
----
Domain
"""
assert domain.id
query = self._find_by_id_query(domain.id)
entity = self._find_first_record(query)
entity.reset(domain.to_dict())
self._session_context.modify(entity)
def remove(self, domain: Domain) -> None:
"""
Remove a domain object.
Parameters:
----
:param domain: domain object.
Returns:
----
Domain
"""
assert domain.id
query = self._find_by_id_query(domain.id)
entity = self._find_first_record(query)
self._session_context.remove(entity)
def find_by_id(self, id: int) -> Domain:
"""
Find a domain object by id.
Parameters:
----
:param id: id of the entity.
Returns:
----
Domain
"""
query = self._find_by_id_query(id)
return self._find_first_domain(query)
def _find_by_id_query(self, id: int) -> Query[Entity]:
"""
Find a domain object by id.
Parameters:
----
:param id: id of the entity.
Returns:
----
Query[Entity]
"""
query = self._entity.query.filter_by(id=id)
return query
def _find_first_domain(self, query: Query[Entity]) -> Domain:
"""
Find the first domain object.
Parameters:
----
:param query: query.
Returns:
----
Domain
"""
record = self._find_first_record(query)
domain = self._get_domain_from_entity(record)
return domain
def _find_all(self, query: Query[Entity]) -> list[Domain]:
"""
Find all domain objects.
Parameters:
----
:param query: query.
Returns:
----
list[Domain]
"""
records = self._find_all_records(query)
return [
self._get_domain_from_entity(record) for record in records
]
def _find_first_record(self, query: Query[Entity]) -> Entity:
"""
Find the first user record.
Parameters:
----
:param query: query.
Returns:
----
Entity
"""
record = query.first()
if not record:
raise self._exception()
return record
def _find_all_records(self, query: Query[Entity]) -> list[Entity]:
"""
Find all records.
Parameters:
----
:param query: query.
Returns:
----
list[Entity]
"""
records = query.all()
return records
def _get_domain_from_entity(self, entity: Entity) -> Domain:
"""
Get the entity to domain.
Parameters:
----
:param entity: entity used.
Returns:
----
Domain
"""
return self._domain.from_dict(entity.to_dict())
def _get_entity_from_domain(self, domain: Domain) -> Entity:
"""
Get the domain to entity.
Parameters:
----
:param domain: domain used.
Returns:
----
Entity
"""
return self._entity(**domain.to_dict())
Обновлено: я хочу, чтобы mypy вызывал такие ошибки: Argument 1 to "reset" of "UserModel" has incompatible type "TermsOfUseDomainDict"; expected "UserDomainDict" [arg-type] mypy(error) при создании экземпляра шаблона репозитория следующим образом:
repo = RepoPattern(
UserModel,
TermsOfUse,
UserNotFoundException,
SessionManager()
)
Между тем я не могу связать свои общие типы с помощью простого союза как:
entity_bound = Union[UserModel, TermsOfUseModel]
domain_bound = Union[User, TermsOfUse]
Что мне нужно, так это выбрать тип домена, связанный с типом объекта. Например: при выборе типа UserModel в качестве типа сущности. Я бы автоматически выбрал тип домена пользователя в репозитории.
Здравствуйте, спасибо за ваш комментарий. Я только что отредактировал свой текст.
Разве моего ответа недостаточно?






Если вы хотите наследовать от RepoPattern для каждой пары объект-домен, вы можете специализировать TypeVar в производном классе. Кроме того, не следует жестко кодировать entity_bound и domain_bound в модуле абстрактного базового класса и рассмотрите возможность работы с чем-то вроде абстрактных Entity и Domain классов. Я также не понял, почему вы сделали переменную типа SpecificException, поскольку она вообще не отображается в интерфейсе класса. Имея в виду эту идею, код может быть примерно таким:
from typing import TypeVar, Generic, Type
from entity import Entity
from domain import Domain
TEntity = TypeVar("TEntity", bound=Entity)
TDomain = TypeVar("TDomain", bound=Domain)
class RepositoryBase(Generic[TEntity, TDomain]):
def __init__(
self, entity_type: Type[TEntity], domain_type: Type[TDomain]
) -> None:
...
class UserRepository(RepositoryBase[UserModel, User]):
...
Но если вы предпочитаете сохранить свое решение, вы можете overload функцию __init__ принимать только определенные комбинации:
from __future__ import annotations
from typing import TypeVar, Generic, Type, overload
from entity import Entity
from domain import Domain
TEntity = TypeVar("TEntity", bound=Entity)
TDomain = TypeVar("TDomain", bound=Domain)
class Repository(Generic[TEntity, TDomain]):
@overload
def __init__(
self: Repository[UserModel, User],
entity_type: Type[UserModel],
domain_type: Type[User],
) -> None:
pass
@overload
def __init__(
self: Repository[TermOfUseModel, TermOfUse],
entity_type: Type[TermOfUseModel],
domain_type: Type[TermOfUse],
) -> None:
pass
def __init__(self, entity_type, domain_type) -> None:
...
Repository(UserModel, User)
Repository(TermOfUseModel, TermOfUse)
Repository(
UserModel, TermOfUse
) # error: Argument 2 to "Repository" has incompatible type "type[TermOfUse]"; expected "type[User]" [arg-type]
Вам следует избегать аннотирования фактической реализации __init__, потому что объединение двух перегрузок не охватывает все комбинации. Вместо этого аннотируйте переменную self в перегрузках, чтобы помочь Mypy в выводе TypeVars.
Некоторые примечания:
src, так как ваш импорт не будет работать, когда код установлен.Entity в вашем коде. Попробуйте удалить его из переменных типа и использовать сопоставление или функцию среды выполнения для создания или работы с сущностями из домена или типа домена.Большое спасибо за ваш ответ. Я знал, что то, что я хотел сделать, было немного растянутым, но на мгновение надеялся, что mypy может помочь. В любом случае, я буду придерживаться своего шаблонного кода вместо использования автомапперов, так как у меня не так много свойств, которые нужно установить.
Не могли бы вы привести пример, который вы хотите предотвратить?