Repositories
Concrete implementations of repository interfaces. A repository takes entities and returns entities, while hiding storage details. It can work against local, remote, data services or third party services.
from typing import Any
from common.exceptions import NotFoundException
from config import config
from data_providers.clients.client_interface import ClientInterface
from data_providers.clients.mongodb.mongo_database_client import MongoDatabaseClient
from features.todo.entities.todo_item import TodoItem
from features.todo.repository.todo_repository_interface import TodoRepositoryInterface
def to_dict(todo_item: TodoItem) -> dict[str, Any]:
_dict: dict[str, Any] = todo_item.__dict__
_dict["_id"] = todo_item.id
return _dict
def get_todo_repository() -> "TodoRepository":
mongo_database_client = MongoDatabaseClient(collection_name="todos", database_name=config.MONGODB_DATABASE)
return TodoRepository(client=mongo_database_client)
class TodoRepository(TodoRepositoryInterface):
client: ClientInterface
def __init__(self, client: ClientInterface):
self.client = client
def update(self, todo_item: TodoItem) -> TodoItem:
updated_todo_item = self.client.update(todo_item.id, to_dict(todo_item))
return TodoItem.from_dict(updated_todo_item)
def delete(self, todo_item_id: str) -> None:
is_deleted = self.client.delete(todo_item_id)
if not is_deleted:
raise NotFoundException
def delete_all(self) -> None:
self.client.delete_collection()
def get(self, todo_item_id: str) -> TodoItem:
todo_item = self.client.get(todo_item_id)
return TodoItem.from_dict(todo_item)
def create(self, todo_item: TodoItem) -> TodoItem | None:
inserted_todo_item = self.client.create(to_dict(todo_item))
return TodoItem.from_dict(inserted_todo_item)
def get_all(self) -> list[TodoItem]:
todo_items: list[TodoItem] = []
for item in self.client.list_collection():
todo_items.append(TodoItem.from_dict(item))
return todo_items
def find_one(self, filter: dict[str, Any]) -> TodoItem | None:
todo_item = self.client.find_one(filter)
if todo_item:
return TodoItem.from_dict(todo_item)
return None
Testing repositories
Use the test_client fixture as input to TodoRepository. The test_client fixture are using the mongomock instead of
real database.
import pytest
from common.exceptions import NotFoundException, ValidationException
from data_providers.clients.mongodb.mongo_database_client import MongoDatabaseClient
from features.todo.entities.todo_item import TodoItem
from features.todo.repository.todo_repository import TodoRepository
class TestTodoRepository:
@pytest.fixture(autouse=True)
def _setup_repository(self, test_client: MongoDatabaseClient):
self.repository = TodoRepository(client=test_client)
def test_create(self):
todo_item = TodoItem(id="1234", title="todo 1", user_id="xyz")
self.repository.create(todo_item)
assert len(self.repository.get_all()) == 1
def test_create_already_exists(self):
todo_item_1 = TodoItem(id="1234", title="todo 1", user_id="xyz")
self.repository.create(todo_item_1)
with pytest.raises(ValidationException):
todo_item_2 = TodoItem(id="1234", title="todo 1", user_id="xyz")
self.repository.create(todo_item_2)
def test_find_item_that_exist(self):
documents = [
{"_id": "81549300", "title": "todo 1", "user_id": "xyz"},
{"_id": "1a2b", "title": "todo 2", "user_id": "xyz"},
{"_id": "987321", "title": "todo 3", "user_id": "abc"},
{"_id": "987456", "title": "todo 4", "user_id": "abc"},
]
self.repository.client.insert_many(documents)
todo_item = self.repository.find_one({"title": "todo 2", "user_id": "xyz"})
assert todo_item is not None and todo_item.id == "1a2b"
def test_find_item_that_does_not_exist(self):
documents = [
{"_id": "81549300", "title": "todo 1", "user_id": "xyz"},
{"_id": "1a2b", "title": "todo 2", "user_id": "xyz"},
{"_id": "987321", "title": "todo 3", "user_id": "abc"},
{"_id": "987456", "title": "todo 4", "user_id": "abc"},
]
self.repository.client.insert_many(documents)
assert self.repository.find_one({"_id": "invalid"}) is None
def test_find_item_of_other_user(self):
documents = [
{"_id": "81549300", "title": "todo 1", "user_id": "xyz"},
{"_id": "1a2b", "title": "todo 2", "user_id": "xyz"},
{"_id": "987321", "title": "todo 3", "user_id": "abc"},
{"_id": "987456", "title": "todo 4", "user_id": "abc"},
]
self.repository.client.insert_many(documents)
assert self.repository.find_one({"_id": "1a2b", "user_id": "abc"}) is None
def test_get_item_that_does_exist(self):
documents = [
{"_id": "81549300", "title": "todo 1", "user_id": "xyz"},
{"_id": "1a2b", "title": "todo 2", "user_id": "xyz"},
{"_id": "987321", "title": "todo 3", "user_id": "abc"},
{"_id": "987456", "title": "todo 4", "user_id": "abc"},
]
self.repository.client.insert_many(documents)
assert self.repository.get("987321").id == "987321"
def test_get_item_that_does_not_exist(self):
documents = [
{"_id": "81549300", "title": "todo 1", "user_id": "xyz"},
{"_id": "1a2b", "title": "todo 2", "user_id": "xyz"},
{"_id": "987321", "title": "todo 3", "user_id": "abc"},
{"_id": "987456", "title": "todo 4", "user_id": "abc"},
]
self.repository.client.insert_many(documents)
with pytest.raises(NotFoundException):
self.repository.get("invalid")
def test_update_item(self):
todo_item = TodoItem(id="81549300", title="todo 1", user_id="xyz")
self.repository.create(todo_item)
todo_item_to_update = TodoItem(id="81549300", title="Updated title", user_id="xyz")
self.repository.update(todo_item=todo_item_to_update)
assert self.repository.get("81549300").title == "Updated title"
def test_update_item_that_does_not_exist(self):
todo_item_to_update = TodoItem(id="unknown", title="Updated title", user_id="xyz")
with pytest.raises(NotFoundException):
self.repository.update(todo_item_to_update)
def test_delete(self):
documents = [
{"_id": "81549300", "title": "todo 1", "user_id": "xyz"},
{"_id": "1a2b", "title": "todo 2", "user_id": "xyz"},
]
self.repository.client.insert_many(documents)
assert len(self.repository.get_all()) == 2
self.repository.delete("81549300")
assert len(self.repository.get_all()) == 1
assert self.repository.get_all() == [self.repository.get("1a2b")]
def test_delete_all(self):
documents = [
{"_id": "81549300", "title": "todo 1", "user_id": "xyz"},
{"_id": "1a2b", "title": "todo 2", "user_id": "xyz"},
]
self.repository.client.insert_many(documents)
assert len(self.repository.get_all()) == 2
self.repository.delete_all()
assert len(self.repository.get_all()) == 0