Coverage for arclith / adapters / output / memory / repository.py: 100%

1from typing import Generic, Optional, TypeVar 

2from uuid6 import UUID, uuid7 

3 

4from arclith.domain.models.entity import Entity 

5from arclith.domain.ports.repository import Repository 

6 

7T = TypeVar("T", bound = Entity) 

8 

9 

10class InMemoryRepository(Repository[T], Generic[T]): 

11 def __init__(self) -> None: 

12 self._store: dict[UUID, T] = {} 

13 

14 async def create(self, entity: T) -> T: 

15 self._store[entity.uuid] = entity 

16 return entity 

17 

18 async def read(self, uuid: UUID) -> Optional[T]: 

19 return self._store.get(uuid) 

20 

21 async def update(self, entity: T) -> T: 

22 self._store[entity.uuid] = entity 

23 return entity 

24 

25 async def delete(self, uuid: UUID) -> None: 

26 self._store.pop(uuid, None) 

27 

28 async def find_all(self) -> list[T]: 

29 return [e for e in self._store.values() if not e.is_deleted] 

30 

31 async def find_deleted(self) -> list[T]: 

32 return [e for e in self._store.values() if e.is_deleted] 

33 

34 async def duplicate(self, uuid: UUID) -> T: 

35 entity = self._store.get(uuid) 

36 if entity is None or entity.is_deleted: 

37 raise KeyError(f"Entity with uuid {uuid} not found") 

38 clone = entity.model_copy(update={"uuid": uuid7()}) 

39 self._store[clone.uuid] = clone 

40 return clone