Merge pull request #10 from knucklesuganda/fix/repository_obj_specifi…

…cations_collision

Removed Repository obj and specifications collisions, added new 'only' specification for limiting the fields of the models

assimilator/__init__.py

@@ -21,7 +21,7 @@ def optional_dependencies(error: str = "ignore"):
     import assimilator.alchemy
 
 with optional_dependencies():
-    import assimilator.kafka
+    import assimilator.kafka_
 
 with optional_dependencies():
     import assimilator.redis_ as redis

assimilator/alchemy/database/repository.py

@@ -1,4 +1,4 @@
-from typing import Type, Union, Optional, TypeVar, Collection, Dict
+from typing import Type, Union, Optional, TypeVar, Collection
 
 from sqlalchemy import func, select, update, delete
 from sqlalchemy.orm import Session, Query
@@ -62,6 +62,8 @@ def filter(
         return [result[0] for result in self.session.execute(query)]
 
     def update(self, obj: Optional[AlchemyModelT] = None, *specifications, **update_values) -> None:
+        obj, specifications = self._check_obj_is_specification(obj, specifications)
+
         if specifications:
             if not update_values:
                 raise InvalidQueryError(
@@ -95,6 +97,8 @@ def refresh(self, obj: AlchemyModelT) -> None:
         self.session.refresh(obj)
 
     def delete(self, obj: Optional[AlchemyModelT] = None, *specifications: SpecificationType) -> None:
+        obj, specifications = self._check_obj_is_specification(obj, specifications)
+
         if specifications:
             query: Query = self._apply_specifications(
                 query=self.get_initial_query(delete(self.model)),

assimilator/alchemy/database/specifications.py

@@ -1,6 +1,6 @@
-from typing import Collection, Optional
+from typing import Collection, Optional, Iterable
 
-from sqlalchemy.orm import Query
+from sqlalchemy.orm import Query, load_only
 from sqlalchemy.sql.operators import is_
 from sqlalchemy import column, desc
 
@@ -76,11 +76,17 @@ def alchemy_join(targets: Collection, join_args: Collection[dict], query: Query)
     return query
 
 
+@specification
+def alchemy_only(*only_fields: Iterable[str], query: Query):
+    return query.options(load_only(*(only_field for only_field in only_fields)))
+
+
 class AlchemySpecificationList(SpecificationList):
     filter = alchemy_filter
     order = alchemy_order
     paginate = alchemy_paginate
     join = alchemy_join
+    only = alchemy_only
 
 
 __all__ = [

assimilator/core/database/repository.py

@@ -1,7 +1,7 @@
 from functools import wraps
 from abc import ABC, abstractmethod
 from typing import TypeVar, Callable, Generic, final, \
-    Union, Optional, Iterable, Type, Collection
+    Union, Optional, Iterable, Type, Collection, Tuple
 
 from assimilator.core.patterns import ErrorWrapper
 from assimilator.core.patterns.lazy_command import LazyCommand
@@ -53,6 +53,22 @@ def __init__(
         self.refresh = self.error_wrapper.decorate(self.refresh)
         self.count = self.error_wrapper.decorate(self.count)
 
+    @final
+    def _check_obj_is_specification(
+        self,
+        obj: ModelT,
+        specifications: Iterable[SpecificationType]
+    ) -> Tuple[Optional[ModelT], Iterable[SpecificationType]]:
+        """
+        This function is called for parts of the code that use both obj and *specifications.
+        We check that if the obj is a model
+        """
+
+        if not isinstance(obj, self.model) and (obj is not None):
+            return None, (obj, *specifications)     # obj is specification
+
+        return obj, specifications
+
     @property
     def specs(self) -> Type[SpecificationList]:
         """ That property is used to shorten the full name of the self.specifications. """

assimilator/core/database/specifications.py

@@ -47,6 +47,7 @@ class SpecificationList:
     order: SpecificationType
     paginate: SpecificationType
     join: SpecificationType
+    only: SpecificationType
 
 
 __all__ = [

assimilator/core/patterns/error_wrapper.py

@@ -13,8 +13,12 @@ def __init__(
         self.default_error = default_error
 
         self.skipped_errors = skipped_errors or set()
-        self.skipped_errors = set(self.skipped_errors)
-        self.skipped_errors.add(BaseException)
+        self.skipped_errors = {
+            *self.skipped_errors,
+            KeyboardInterrupt,
+            SystemExit,
+            *self.error_mappings.values()  # we want to skip all the mapped values as they are already fixed
+        }
 
     def __enter__(self):
         return self
@@ -43,5 +47,8 @@ def wrapper(*args, **kwargs):
 
         return wrapper
 
+    def __str__(self):
+        return f"{type(self).__name__}({self.error_mappings})"
+
 
 __all__ = ['ErrorWrapper']

assimilator/internal/database/repository.py

@@ -3,8 +3,13 @@
 from assimilator.internal.database.models import InternalModel
 from assimilator.core.patterns.error_wrapper import ErrorWrapper
 from assimilator.internal.database.error_wrapper import InternalErrorWrapper
-from assimilator.core.database import Repository, SpecificationType, LazyCommand, \
-    make_lazy, InvalidQueryError
+from assimilator.core.database import (
+    Repository,
+    SpecificationType,
+    LazyCommand,
+    make_lazy,
+    InvalidQueryError,
+)
 from assimilator.internal.database.specifications import InternalSpecificationList
 
 ModelT = TypeVar("ModelT", bound=InternalModel)
@@ -63,15 +68,22 @@ def save(self, obj: Optional[ModelT] = None, **obj_data) -> ModelT:
         return obj
 
     def delete(self, obj: Optional[ModelT] = None, *specifications: SpecificationType) -> None:
+        obj, specifications = self._check_obj_is_specification(obj, specifications)
+
         if specifications:
             for model in self.filter(*specifications, lazy=True):
                 del self.session[model.id]
         elif obj is not None:
             del self.session[obj.id]
 
     def update(
-        self, obj: Optional[ModelT] = None, *specifications: SpecificationType, **update_values,
+        self,
+        obj: Optional[ModelT] = None,
+        *specifications: SpecificationType,
+        **update_values,
     ) -> None:
+        obj, specifications = self._check_obj_is_specification(obj, specifications)
+
         if specifications:
             if not update_values:
                 raise InvalidQueryError(
@@ -95,7 +107,9 @@ def refresh(self, obj: ModelT) -> None:
 
     @make_lazy
     def count(
-        self, *specifications: SpecificationType, lazy: bool = False,
+        self,
+        *specifications: SpecificationType,
+        lazy: bool = False,
     ) -> Union[LazyCommand[int], int]:
         if specifications:
             return len(self.filter(*specifications, lazy=False))

assimilator/internal/database/specifications.py

@@ -1,5 +1,5 @@
 import operator
-from typing import List, Iterable, Any, Union, Callable, Optional
+from typing import List, Iterable, Any, Union, Callable, Optional, Set
 
 from assimilator.core.database import specification, SpecificationList, filter_parameter_parser
 from assimilator.internal.database.models import InternalModel
@@ -86,11 +86,21 @@ def internal_join(*args, query: QueryT, **kwargs) -> Any:
     return query
 
 
+@specification
+def internal_only(*only_fields: Iterable[str], query: QueryT):
+    if isinstance(query, str):
+        return query
+
+    only_fields = set(only_fields)
+    return [model.copy(include=only_fields) for model in query]
+
+
 class InternalSpecificationList(SpecificationList):
     filter = internal_filter
     order = internal_order
     paginate = internal_paginate
     join = internal_join
+    only = internal_only
 
 
 __all__ = [

assimilator/kafka_/__init__.py

@@ -0,0 +1 @@
+from assimilator.kafka_.events.events_bus import *

assimilator/kafka/events/events_bus.py → assimilator/kafka_/events/events_bus.py

@@ -1,8 +1,8 @@
 import json
 from typing import Iterable
 
-from kafka import KafkaConsumer, KafkaProducer
-from kafka.errors import KafkaError
+from kafka_ import KafkaConsumer, KafkaProducer
+from kafka_.errors import KafkaError
 
 from assimilator.core.events import Event, ExternalEvent
 from assimilator.core.events import EventParsingError, EventProducingError

assimilator/redis_/database/repository.py

@@ -4,16 +4,12 @@
 from redis.client import Pipeline
 
 from assimilator.redis_.database import RedisModel
-from assimilator.core.database.exceptions import DataLayerError, NotFoundError, InvalidQueryError
 from assimilator.core.patterns.error_wrapper import ErrorWrapper
 from assimilator.core.database import (
-    SpecificationList,
-    SpecificationType,
-    Repository,
-    LazyCommand,
-    make_lazy,
+    SpecificationList, SpecificationType, Repository, LazyCommand, make_lazy,
 )
 from assimilator.internal.database.specifications import InternalSpecificationList
+from assimilator.core.database.exceptions import DataLayerError, NotFoundError, InvalidQueryError
 
 RedisModelT = TypeVar("RedisModelT", bound=RedisModel)
 
@@ -101,8 +97,10 @@ def save(self, obj: Optional[RedisModelT] = None, **obj_data) -> RedisModelT:
         return obj
 
     def delete(self, obj: Optional[RedisModelT] = None, *specifications: SpecificationType) -> None:
+        obj, specifications = self._check_obj_is_specification(obj, specifications)
+
         if specifications:
-            models = self.filter(*specifications)   # TODO: ADD ONLY SPECIFICATIONS
+            models = self.filter(*specifications)
             self.transaction.delete(*[str(model.id) for model in models])
         elif obj is not None:
             self.transaction.delete(obj.id)
@@ -113,6 +111,8 @@ def update(
         *specifications: SpecificationType,
         **update_values,
     ) -> None:
+        obj, specifications = self._check_obj_is_specification(obj, specifications)
+
         if specifications:
             if not update_values:
                 raise InvalidQueryError(

examples/alchemy/database/main.py

@@ -76,10 +76,21 @@ def order_users(repository: Repository):
     for i, ordered_user in enumerate(repository.filter(
         repository.specs.order('id', '-balance'),
         repository.specs.paginate(offset=20, limit=40),
+        repository.specs.only('username', 'id', 'balance'),
     )):
         print(f"User {i} ordered by id and balance:", ordered_user.username, ordered_user.balance)
 
 
+def update_poor_users(uow: UnitOfWork):
+    with uow:
+        uow.repository.update(
+            uow.repository.specs.filter(balance__lt=50),
+            balance=100,
+        )
+
+        uow.commit()
+
+
 if __name__ == '__main__':
     new_user_id = create_user(
         username="Andrey",
@@ -110,3 +121,6 @@ def order_users(repository: Repository):
         print("User was not found due to his deletion! Error:", error)
 
     order_users(repository=create_uow().repository)
+
+    update_poor_users(create_uow())
+    order_users(repository=create_uow().repository)

examples/alchemy/database/models.py

@@ -2,7 +2,7 @@
 from sqlalchemy.orm import declarative_base, sessionmaker
 
 
-engine = create_engine(url="sqlite:///:memory:")
+engine = create_engine(url="sqlite:///:memory:", echo="debug")
 DatabaseSession = sessionmaker(bind=engine)
 Base = declarative_base()
 

examples/redis/database/main.py

@@ -50,6 +50,7 @@ def show_rich_users(balance: int, repository: Repository):
     users: LazyCommand[RedisUser] = repository.filter(
         repository.specs.filter(balance__gt=balance),
         repository.specs.paginate(limit=10),
+        repository.specs.only('username', 'balance'),
         lazy=True,
     )
 

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='py_assimilator',
-    version='0.3.0',
+    version='0.3.1',
     author='Andrey Ivanov',
     author_email='[email protected]',
     url='https://pypi.python.org/pypi/py_assimilator/',