在Python中链接类型边界对。

huangapple go评论58阅读模式
英文:

Link pairs of type bounds together in Python

问题

我正在尝试创建一个抽象的仓库类,以便可以大幅减少我的样板代码的长度。同时,为了实现这一目标,我需要将Entity的边界链接到特定的Domain(类似于在 TypeScript 中使用 Pick)以引起 mypy 的关注,例如Missing positional arguments。在 Python 中是否有可能这样做?

"""
通用仓库
"""
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]):
    """
    仓库模式
    """

    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:
        """
        添加新的领域对象。

        参数:
        ----
            :param domain: 领域对象。

        返回:
        ----
            Domain
        """
        entity = self._get_entity_from_domain(domain)
        self._session_context.add(entity)

    def modify(self, domain: Domain) -> None:
        """
        添加新的领域对象。

        参数:
        ----
            :param domain: 领域对象。

        返回:
        ----
            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:
        """
        删除领域对象。

        参数:
        ----
            :param domain: 领域对象。

        返回:
        ----
            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:
        """
        通过ID查找领域对象。

        参数:
        ----
            :param id: 实体的ID。

        返回:
        ----
            Domain
        """
        query = self._find_by_id_query(id)
        return self._find_first_domain(query)

    def _find_by_id_query(self, id: int) -> Query[Entity]:
        """
        通过ID查找领域对象。

        参数:
        ----
            :param id: 实体的ID。

        返回:
        ----
            Query[Entity]
        """
        query = self._entity.query.filter_by(id=id)
        return query

    def _find_first_domain(self, query: Query[Entity]) -> Domain:
        """
        查找第一个领域对象。

        参数:
        ----
            :param query: 查询。

        返回:
        ----
            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]:
        """
        查找所有领域对象。

        参数:
        ----
            :param query: 查询。

        返回:
        ----
            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:
        """
        查找第一个用户记录。

        参数:
        ----
            :param query: 查询。

        返回:
        ----
            Entity
        """

        record = query.first()

        if not record:
            raise self._exception()

        return record

    def _find_all_records(self, query: Query[Entity]) -> list[Entity]:
        """
        查找所有记录。

        参数:
        ----
            :param query: 查询。

        返回:
        ----
            list[Entity]
        """

        records = query.all()
        return records

    def _get_domain_from_entity(self, entity: Entity) -> Domain:
        """
        从实体获取领域。

        参数:
        ----
            :param entity: 使用的实体。

        返回:
        ----
            Domain
        """
        return self._domain.from_dict(entity.to_dict())

    def _get_entity_from_domain(self, domain: Domain) -> Entity:
        """
        从领域获取实体。

        参数:
        ----
            :param domain: 使用的领域。

        返回:
        ----
            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()
)

与此同时,我不能使用简单的 Union 来限定我的泛型类型,如下所示:

entity_bound = Union[UserModel, TermsOfUseModel]
domain_bound = Union[User, TermsOfUse]

我需要根据实体类型选择一个领域类型。例如:当选择 UserModel 类型作为实体类型时,我希望自动选择 User 领域类型在仓库中。

英文:

I am trying to create an abstract repository class so that I can reduce drastically the length of my boilerplate code. Meanwhile, to do so, I would need to link the bounds of Entity to a certain Domain (Like when using Pick in Typescript) in order to raise mypy attention such Missing positional arguments. Is it possible to do so in 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())

Edit: I want mypy to raise these kind of errors: Argument 1 to "reset" of "UserModel" has incompatible type "TermsOfUseDomainDict"; expected "UserDomainDict" [arg-type] mypy(error) while instantiating the repository pattern like that:

repo = RepoPattern(
UserModel,
TermsOfUse,
UserNotFoundException,
SessionManager()
)

Meanwhile I cannot bound my generic types using a simple Union as:

entity_bound = Union[UserModel, TermsOfUseModel]
domain_bound = Union[User, TermsOfUse]

What I need is picking a domain type bound given an entity type.
For instance: When selecting a UserModel type as an entity type. I would automatically select a User domain type in the repository.

答案1

得分: 1

如果你想要为每个实体-领域对继承RepoPattern,你可以在派生类中专门化TypeVars。另外,不要在抽象基类模块中硬编码entity_bounddomain_bound,考虑使用抽象的EntityDomain类。我也不明白为什么你创建了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类型变量的公共接口。考虑将其从类型变量中移除,并使用运行时映射或函数来创建或处理来自领域或领域类型的实体。
  • 在不同层之间的类似或相关对象之间的映射不是存储库的责任。考虑使用自动映射器,如py-automapperobject-mapper。或者,你可以实现自己的特定映射器或将映射逻辑放在外层的类中。
  • Mypy本身不支持Type mappings
  • 或者,你可能会发现Active Record模式及其在Python中的实现更适合你的需求。
英文:

If you want to inherit from the RepoPattern for each entity-domain pair, you can specialize the TypeVars in the derived class. Also, do not hard code the entity_bound and domain_bound in the abstract base class module, and consider working with something like abstract Entity and Domain classes. I also didn't understand why you made the SpecificException type variable since it doesn't appear in the class interface at all. With this idea in mind, the code could be something similar to this:

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]):
...

But if you prefer to keep your solution, you can overload the __init__ function to only accept specific combinations:

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]

You should avoid annotating the actual __init__ implementation because the union of two overloads does not cover all combinations. Instead, annotate the self variable in the overloads to guide Mypy in deducing TypeVars.

Some side notes:

  • Avoid importing modules from the src directory, as your imports will not work when the code is installed.
  • I couldn't find any public interface with the Entity type variable usage in your code. Consider removing it from the type variables and using a runtime mapping or function to create or work with entities from the domain or domain type.
  • Mapping between similar or related objects in different layers is not the repository's responsibility. Consider using auto-mappers like py-automapper or object-mapper. Alternatively, you can implement your specific mappers or place the mapping logic in the class on the outer layer.
  • Mypy itself doesn't support Type mappings.
  • Alternatively, you might find the Active Record pattern and its implementations in Python more suitable for your needs.

huangapple
  • 本文由 发表于 2023年7月27日 20:41:16
  • 转载请务必保留本文链接:https://go.coder-hub.com/76779836.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定