repository.py 3.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. # (c) Nelen & Schuurmans
  2. from typing import Any
  3. from typing import Generic
  4. from typing import List
  5. from typing import Optional
  6. from typing import Type
  7. from typing import TypeVar
  8. from typing import Union
  9. from .exceptions import DoesNotExist
  10. from .filter import Filter
  11. from .gateway import Gateway
  12. from .pagination import Page
  13. from .pagination import PageOptions
  14. from .root_entity import RootEntity
  15. from .types import Id
  16. from .types import Json
  17. __all__ = ["Repository"]
  18. T = TypeVar("T", bound=RootEntity)
  19. class Repository(Generic[T]):
  20. entity: Type[T]
  21. def __init__(self, gateway: Gateway):
  22. self.gateway = gateway
  23. def __init_subclass__(cls) -> None:
  24. (base,) = cls.__orig_bases__ # type: ignore
  25. (entity,) = base.__args__
  26. assert issubclass(entity, RootEntity)
  27. super().__init_subclass__()
  28. cls.entity = entity
  29. async def all(self, params: Optional[PageOptions] = None) -> Page[T]:
  30. return await self.filter([], params=params)
  31. async def by(
  32. self, key: str, value: Any, params: Optional[PageOptions] = None
  33. ) -> Page[T]:
  34. return await self.filter([Filter(field=key, values=[value])], params=params)
  35. async def filter(
  36. self, filters: List[Filter], params: Optional[PageOptions] = None
  37. ) -> Page[T]:
  38. records = await self.gateway.filter(filters, params=params)
  39. total = len(records)
  40. # when using pagination, we may need to do a count in the db
  41. # except in a typical 'first page' situation with few records
  42. if params is not None and not (params.offset == 0 and total < params.limit):
  43. total = await self.count(filters)
  44. return Page(
  45. total=total,
  46. limit=params.limit if params else None,
  47. offset=params.offset if params else None,
  48. items=[self.entity(**x) for x in records],
  49. )
  50. async def get(self, id: Id) -> T:
  51. res = await self.gateway.get(id)
  52. if res is None:
  53. raise DoesNotExist("object", id)
  54. else:
  55. return self.entity(**res)
  56. async def add(self, item: Union[T, Json]) -> T:
  57. if isinstance(item, dict):
  58. item = self.entity.create(**item)
  59. created = await self.gateway.add(item.model_dump())
  60. return self.entity(**created)
  61. async def update(self, id: Id, values: Json) -> T:
  62. if not values:
  63. return await self.get(id)
  64. updated = await self.gateway.update_transactional(
  65. id, lambda x: self.entity(**x).update(**values).model_dump()
  66. )
  67. return self.entity(**updated)
  68. async def upsert(self, item: T) -> T:
  69. values = item.model_dump()
  70. upserted = await self.gateway.upsert(values)
  71. return self.entity(**upserted)
  72. async def remove(self, id: Id) -> bool:
  73. return await self.gateway.remove(id)
  74. async def count(self, filters: List[Filter]) -> int:
  75. return await self.gateway.count(filters)
  76. async def exists(self, filters: List[Filter]) -> bool:
  77. return await self.gateway.exists(filters)