-
Notifications
You must be signed in to change notification settings - Fork 1
/
abstract.py
163 lines (126 loc) · 4.96 KB
/
abstract.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
"""Define the interface of the repositories."""
import abc
from typing import Dict, List, Type, TypeVar, Union
from ..exceptions import EntityNotFoundError
from ..model import Entity as EntityModel
Entity = TypeVar("Entity", bound=EntityModel)
class AbstractRepository(abc.ABC):
"""Gather common methods and define the interface of the repositories.
Attributes:
database_url: URL specifying the connection to the database.
"""
@abc.abstractmethod
def __init__(self, database_url: str = "") -> None:
"""Initialize the repository attributes.
Args:
database_url: URL specifying the connection to the database.
"""
self.database_url = database_url
@abc.abstractmethod
def add(self, entity: Entity) -> None:
"""Append an entity to the repository.
Args:
entity: Entity to add to the repository.
"""
# no cover: it's tested by it's subclasses
if entity.id_ < 0: # pragma: no cover
entity.id_ = self._next_id(entity) # pragma: no cover
raise NotImplementedError
@abc.abstractmethod
def delete(self, entity: Entity) -> None:
"""Delete an entity from the repository.
Args:
entity: Entity to remove from the repository.
"""
raise NotImplementedError
@abc.abstractmethod
def get(self, entity_model: Type[Entity], entity_id: Union[str, int]) -> Entity:
"""Obtain an entity from the repository by it's ID.
Args:
entity_model: Type of entity object to obtain.
entity_id: ID of the entity object to obtain.
Returns:
entity: Entity object that matches the search criteria.
Raises:
EntityNotFoundError: If the entity is not found.
"""
raise NotImplementedError
@abc.abstractmethod
def all(self, entity_model: Type[Entity]) -> List[Entity]:
"""Obtain all the entities of a type from the repository.
Args:
entity_model: Type of entity objects to obtain.
Returns:
entities: List of Entity object that matches the search criteria.
Raises:
EntityNotFoundError: If the entities are not found.
"""
raise NotImplementedError
@abc.abstractmethod
def commit(self) -> None:
"""Persist the changes into the repository."""
raise NotImplementedError
@abc.abstractmethod
def search(
self, entity_model: Type[Entity], fields: Dict[str, Union[str, int]]
) -> List[Entity]:
"""Obtain the entities whose attributes match one or several conditions.
Args:
entity_model: Type of entity object to obtain.
fields: Dictionary with the {key}:{value} to search.
Returns:
entities: List of Entity object that matches the search criteria.
Raises:
EntityNotFoundError: If the entities are not found.
"""
raise NotImplementedError
@abc.abstractmethod
def apply_migrations(self, migrations_directory: str) -> None:
"""Run the migrations of the repository schema.
Args:
migrations_directory: path to the directory containing the migration
scripts.
"""
raise NotImplementedError
def last(self, entity_model: Type[Entity]) -> Entity:
"""Get the biggest entity from the repository.
Args:
entity_model: Type of entity object to obtain.
Returns:
entity: Biggest Entity object of type entity_model.
Raises:
EntityNotFoundError: If there are no entities.
"""
try:
return max(self.all(entity_model))
except KeyError as error:
# no cover: it's tested by it's subclasses
raise EntityNotFoundError( # pragma: no cover
f"There are no {entity_model.__name__}s in the repository."
) from error
def first(self, entity_model: Type[Entity]) -> Entity:
"""Get the smallest entity from the repository.
Args:
entity_model: Type of entity object to obtain.
Returns:
entity: Smallest Entity object of type entity_model.
Raises:
EntityNotFoundError: If there are no entities.
"""
try:
return min(self.all(entity_model))
except KeyError as error:
# no cover: it's tested by it's subclasses
raise EntityNotFoundError( # pragma: no cover
f"There are no {entity_model.__name__}s in the repository."
) from error
def _next_id(self, entity: Entity) -> int:
"""Return one id unit more than the last entity id in the repository.
Args:
entity: Entity whose model we want to get the next entity id.
"""
try:
last_id = self.last(type(entity)).id_
except EntityNotFoundError:
return 0
return last_id + 1