Merge pull request #30 from knucklesuganda/feature/usability

✨Usability Update(1.3.0)✨

assimilator/alchemy/__init__.py

@@ -1,7 +1,19 @@
 import sqlalchemy
 
+from assimilator.core.services import CRUDService
+from assimilator.core.usability.registry import register_provider, PatternList
+from assimilator.alchemy.database import AlchemyRepository, AlchemyUnitOfWork
+
+
 if sqlalchemy.__version__ < "2.0.0":
     raise RuntimeWarning(
         "PyAssimilator will only support SQLAlchemy 2 from now on. Please, update "
         "the library using this manual: https://docs.sqlalchemy.org/en/20/changelog/migration_20.html"
     )
+
+
+register_provider(provider='alchemy', pattern_list=PatternList(
+    repository=AlchemyRepository,
+    uow=AlchemyUnitOfWork,
+    crud=CRUDService
+))

assimilator/core/usability/exceptions.py

@@ -0,0 +1,8 @@
+
+
+class PatternNotFoundError(KeyError):
+    pass
+
+
+class ProviderNotFoundError(KeyError):
+    pass

assimilator/core/usability/pattern_creator.py

@@ -0,0 +1,52 @@
+from typing import TypeVar, Any, Type, Dict
+
+from assimilator.core.usability.registry import get_pattern
+from assimilator.core.database import Repository, UnitOfWork
+from assimilator.core.services import CRUDService
+
+ModelT = TypeVar("ModelT")
+
+
+def create_repository(
+    provider: str,
+    model: Type[ModelT],
+    session: Any,
+    kwargs_repository: Dict[str, Any] = None,
+) -> Repository:
+    repository_cls: Type[Repository] = get_pattern(provider=provider, pattern_name='repository')
+    return repository_cls(model=model, session=session, **(kwargs_repository or {}))
+
+
+def create_uow(
+    provider: str,
+    model: Type[ModelT],
+    session: Any,
+    kwargs_repository: Dict[str, Any] = None,
+    kwargs_uow: Dict[str, Any] = None,
+) -> UnitOfWork:
+    repository = create_repository(
+        provider=provider,
+        model=model,
+        session=session,
+        kwargs_repository=kwargs_repository,
+    )
+    uow_cls: Type[UnitOfWork] = get_pattern(provider=provider, pattern_name='uow')
+    return uow_cls(repository=repository, **(kwargs_uow or {}))
+
+
+def create_crud(
+    provider: str,
+    model: Type[ModelT],
+    session: Any,
+    kwargs_repository: Dict[str, Any] = None,
+    kwargs_uow: Dict[str, Any] = None,
+) -> CRUDService:
+    uow = create_uow(
+        provider=provider,
+        model=model,
+        session=session,
+        kwargs_repository=kwargs_repository,
+        kwargs_uow=kwargs_uow,
+    )
+    crud_cls: Type[CRUDService] = get_pattern(provider=provider, pattern_name='crud')
+    return crud_cls(uow=uow)

assimilator/core/usability/registry.py

@@ -0,0 +1,62 @@
+import importlib
+from typing import Dict, Type, Union
+
+from pydantic import BaseModel
+
+from assimilator.core.services.crud import CRUDService
+from assimilator.core.database import Repository, UnitOfWork
+from assimilator.core.usability.exceptions import PatternNotFoundError, ProviderNotFoundError
+
+
+class PatternList(BaseModel):
+    class Config:
+        frozen = True
+
+    repository: Type[Repository]
+    uow: Type[UnitOfWork]
+    crud: Type[CRUDService]
+
+
+registry: Dict[str, PatternList] = {}
+
+
+def register_provider(provider: str, pattern_list: PatternList):
+    registry[provider] = pattern_list
+
+
+def find_provider(provider_path: str):
+    """ Imports a module that has automatic pattern registration """
+    importlib.import_module(provider_path)
+
+
+def get_pattern_list(provider: str):
+    return registry[provider]
+
+
+def unregister_provider(provider: str):
+    try:
+        del registry[provider]
+    except KeyError:
+        raise ProviderNotFoundError(f"Provider {provider} was not found")
+
+
+def get_pattern(provider: str, pattern_name: str) -> Type[Union[Repository, UnitOfWork, CRUDService]]:
+    try:
+        pattern_cls = getattr(registry[provider], pattern_name, None)
+    except KeyError:
+        raise ProviderNotFoundError(f"Provider {pattern_name} was not found")
+
+    if pattern_cls is None:
+        raise PatternNotFoundError(f"Pattern '{pattern_name}' for {provider} provider was not found")
+
+    return pattern_cls
+
+
+__all__ = [
+    'register_provider',
+    'unregister_provider',
+    'PatternList',
+    'get_pattern_list',
+    'get_pattern',
+    'find_provider',
+]

assimilator/internal/__init__.py

@@ -0,0 +1,9 @@
+from assimilator.core.services import CRUDService
+from assimilator.core.usability.registry import register_provider, PatternList
+from assimilator.internal.database import InternalRepository, InternalUnitOfWork
+
+register_provider(provider='internal', pattern_list=PatternList(
+    repository=InternalRepository,
+    uow=InternalUnitOfWork,
+    crud=CRUDService,
+))

assimilator/mongo/__init__.py

@@ -0,0 +1,9 @@
+from assimilator.core.services import CRUDService
+from assimilator.core.usability.registry import register_provider, PatternList
+from assimilator.mongo.database import MongoRepository, MongoUnitOfWork
+
+register_provider(provider='mongo', pattern_list=PatternList(
+    repository=MongoRepository,
+    uow=MongoUnitOfWork,
+    crud=CRUDService,
+))

assimilator/redis_/__init__.py

@@ -0,0 +1,12 @@
+from assimilator.core.services import CRUDService
+from assimilator.redis_.database import RedisRepository, RedisUnitOfWork
+from assimilator.core.usability.registry import register_provider, PatternList
+
+pattern_list = PatternList(
+    repository=RedisRepository,
+    uow=RedisUnitOfWork,
+    crud=CRUDService,
+)
+
+register_provider(provider='redis', pattern_list=pattern_list)
+register_provider(provider='redis_', pattern_list=pattern_list)