In questo post continueremo l’esplorazione del nostro progetto open source, per capire come eseguire operazioni CRUD (Create, Read, Update, Delete) sui record del database.
Nel nostro post precedente, abbiamo iniziato ad esplorare la struttura della nostra classe CRUDHandler
Create: Aggiungere Record
La prima fra le operazioni CRUD è la creazione di nuovi record del database, gestita dal metodo CRUDHandler.add()ExpenseAdd
# modules/crud_handler.py
106 def add(self, data: ExpenseAdd):
107 """Add expense to the DB.
108
109 Parameters
110 -----------------------
111 data : ExpenseAdd
112 Expense data.
113 """
114 # Primary key added automatically
115 self.session.add(Expense(**data.model_dump()))
116 self.session.commit()Alla riga 115, trasformo l’ExpenseAddBaseModelExpenseid mancante, poiché è stato impostato come chiave primaria (e quindi verrà inserito automaticamente dal database).
L’oggetto ExpenseSession.add() (che accetta un’instanza del modello di dati come argomento). Il metodo Session.commit() è infine chiamato per applicare in modo permanente la modifica al database.
Read: Cercare Record
Progettare una funzione per l’estrazione di record da un database richiede in primo luogo di decidere quali criteri verranno utilizzati per cercare nel database: in base alla mia esperienza con programmi simili a sem in passato, ho deciso nel modo seguente.
- L’utente dovrebbe essere in grado di selezionare una data iniziale e una finale entro cui le spese restituite dovranno rientrare. Una o entrambe le date dovrebbero essere opzionali: specificando solo una data di inizio o di fine verranno restituite tutte le spese precedenti o successive, rispettivamente, e in assenza di date tutte le spese dovranno essere restituite.
- L’utente dovrebbe essere in grado di filtrare per tipo e categoria. I criteri dovrebbero essere delle liste di valori, nel caso in cui l’utente desideri più di un valore consentito. Dovrebbero anche essere opzionali, non effettuando filtraggio se non vengono passati valori.
Strong Types per Funzioni di Ricerca
Una volta decisi i criteri di ricerca, il primo passo è codificarli in uno strong type QueryParameters, che verrà passato alle funzioni di ricerca come argomento.
# modules/schemas.py
77 class QueryParameters(BaseModel):
78 """Strong type for query parameters.
79
80 Attributes
81 -----------------------
82 start : Optional[datetime.date]
83 Start date (included).
Not filtered on if `None`. Default is `None`.
84 end : Optional[datetime.date]
85 End date (included).
Not filtered on if `None`. Default is `None`.
86 types : Optional[list[str]]
87 Types to filter the query.
If `None`, all types. Default is `None`.
88 categories : Optional[list[str]]
89 Categories to filter the query. If `None`, all types. Default
is
90 `None`.
91 """
92
93 start: Optional[datetime.date] = Field(
94 default=None,
95 description="""Start date (included).
Not filtered on if `None`.
96 Default is `None`.""",
97 )
98 end: Optional[datetime.date] = Field(
99 default=None,
100 description="""End date (included).
Not filtered on if `None`. Default
101 is `None`.""",
102 )
103 types: Optional[list[str]] = Field(
104 default=None,
105 description="""Types to filter the query.
If `None`, all types. Default
106 is `None`.""",
107 )
108 categories: Optional[list[str]] = Field(
109 default=None,
110 description="""Categories to filter the query.
If `None`, all types.
111 Default is `None`.""",
112 )Ho implementato QueryParameters come una classe derivata da BaseModel, per ereditarne le funzionalità di convalida dei campi e serializzazione. La convalida sui campi opzionali è assicurata utilizzando il tipo OptionalNone) definito nel modulo typing. Quando uno dei campi è nullo i relativi criteri di filtraggio non saranno applicati.
Ho anche aggiunto un tipo ExpenseReadExpenseAdd
# modules/schemas.py
115 class ExpenseRead(ExpenseBase):
116 """Type for selection operations.
117
118 Attributes
119 -----------------------
120 id : int
121 ID of the expense, primary key field.
122 """
123
124 id: int = Field(description="ID of the expense, primary key
field.")Ricerca
Ora possiamo esplorare le funzioni di ricerca in CRUDHandler. Durante la sua creazione, ho deciso di avere due funzioni di ricerca separate: una dovrebbe restituire direttamente un elenco di spese, mentre l’altra dovrebbe farne un riassunto (sommando gli importi raggruppati per categoria e tipologia, per riassunti di fine mese e simili).
Queste funzioni condividono gli stessi criteri di ricerca, che si sono rivelati leggermente involuti da costruire. Per evitare ripetizioni, ho creato il metodo privato CRUDHandler._build_query_conditions()QueryParameters
# modules/crud_handler.py
118 def _build_query_conditions(self, params: QueryParameters):
119 if params.start is None:
120 start_condition = True
121 else:
122 start_condition = Expense.date >= params.start
123
124 if params.end is None:
125 end_condition = True
126 else:
127 end_condition = Expense.date <= params.end
128
129 if params.types is None:
130 type_condition = True
131 else:
132 type_condition = Expense.type.in_(params.types)
133
134 if params.categories is None:
135 cat_condition = True
136 else:
137 cat_condition = Expense.category.in_(params.categories)
138
139 return and_(
140 start_condition, end_condition, type_condition,
cat_condition
141 )SQLAlchemy accetta condizioni scritte come espressioni in termini di campi del modello (utilizzando la sintassi <modello>.<campo>). Queste espressioni possono utilizzare operatori SQL, alcuni dei quali sono definiti anche in Python (ad esempio, <= e >=). SQLAlchemy offre anche metodi in grado di riprodurre operatori SQL più complessi (come IN, tramite in_()).
Nel codice sopra, creiamo per ogni campo di QueryParametersand_(), che riproduce l’operatore SQL AND.
La prima funzione di ricerca è CRUDHandlerExpense
# modules/crud_handler.py
143 def query(self, params: QueryParameters) -> list[Expense]:
144 """Return expenses matching specified filters.
145
146 Parameters
147 -----------------------
148 params : QueryParameters
149 Parameters of the query.
150
151 Returns
152 -----------------------
153 list[Expense]
154 List of expenses matching the criteria.
155 """
156 return self.session.scalars(
157 select(Expense)
158 .where(self._build_query_conditions(params))
159 .order_by(Expense.date, Expense.id)
160 ).all()Qui stiamo utilizzando la catena di metodi Session.scalars().all(), che esegue una ricerca e restituisce i risultati come una lista di istanze del tipo di dati del modello (i risultati di ricerca possono anche essere ottenuti in altri formati, usando altri metodi).
L’argomento di questo metodo è un oggetto Select di SQLAlchemy, creato utilizzando come argomenti del costruttore i campi da restituire (o l’intero modello, per richiedere tutti i campi). L’equivalente di clausole SQL come WHERE, GROUP BY (vedi sotto) o ORDER BY può essere invocato come metodi dell’oggetto Select, con condizioni e campi con valori booleani (<modello>.<campo>) come in SQL.
Si noti che il metodo restituisce una lista di Expense, anziché una lista di ExpenseRead (che richiederebbe una conversione esplicita). Questo non sarà un problema, poiché questa funzione verrà chiamata solo dall’API, dove il risultato verrà immediatamente convertito in formato JSON. Le Expense create vengono immediatamente scartate, e non vi è alcuna possibilità di modifiche involontarie al database.
Sommario
Il secondo metodo di ricerca in CRUDHandler è summarize(), che riceve gli stessi criteri di query() ma restituisce le somme del campo amount, raggruppate per categoria e tipo. Il risultato restituito è un dizionario di dizionari, dove le chiavi esterna ed interna sono rispettivamente categoria e tipo:
{
"category 1": {"type 1": <amount 1>, "type 2": <amount 2>, ...},
"category 2": {...},
...
}Questa è l’implementazione:
# modules/crud_handler.py
162 def summarize(
163 self, params: QueryParameters
164 ) -> dict[str, dict[str, float]]:
165 """Summarize expenses matching specified filters.
166
167 Parameters
168 -----------------------
169 params : QueryParameters
170 Parameters of the query.
171
172 Returns
173 -----------------------
174 dict[dict[str, float]]
175 Dictionary containing inner {type: amount} dictionaries,
176 grouped by category.
177 """
178 sums = self.session.execute(
179 select(
180 Expense.category,
181 Expense.type,
182 (func.sum(Expense.amount)).label("sum"),
183 )
184 .where(self._build_query_conditions(params))
185 .group_by(Expense.category, Expense.type)
186 .order_by(Expense.category, Expense.type)
187 ).all()
188
189 res = {}
190
191 # Creating outer dictionaries
192 for s in sums:
193 res[s.category] = {}
194
195 # Creating inner dictionaries
196 for s in sums:
197 res[s.category][s.type] = s.sum
198
199 return resQui sto usando la funzione sum(), che riproduce la funzione di aggregazione SQL SUM(), ed è contenuta nel sottomodulo func di SQLAlchemy, insieme ad altre funzioni applicabili a campi di modelli. Ho anche assegnato un alias al campo sum tramite il metodo label(), disponibile per i campi e le espressioni che coinvolgono campi. Il passaggio finale è la creazione del dizionario da restituire (con qualche ripetizione nella creazione delle chiavi, che ho ritenuto trascurabili).
Update: Aggiornare di Record
Le operazioni CRUD comprendono la possibilità di aggiornare record esistenti, che in sem ho deciso di selezionare in base al loro ID. La funzione che esegue questo compito richiederà i nuovi valori dei campi da modificare (non dovrebbe essere necessario fornire quelli che non cambiano). Ho codificato questi requisiti nello strong type ExpenseUpdate
# modules/schemas.py
127 class ExpenseUpdate(BaseModel):
128 """Container for data to update existing expenses with.
129
130 Attributes set to `None` will not be changed.
131
132 Attributes
133 -----------------------
134 date : Optional[datetime.date]
135 Date of the expense. Default is `None`.
136 type : Optional[str]
137 Low-level group of the expense. Default is `None`.
138 category : Optional[str]
139 High-level group of the expense. Default is `None`.
140 amount : Optional[float]
141 Amount of the expense. Default is `None`.
142 description : Optional[str]
143 Description of the expense. Default is `None`.
144 """
145
146 date: Optional[datetime.date] = Field(
147 default=None,
148 description="Date of the expense.",
149 )
150 type: Optional[str] = Field(
151 default=None,
152 description="Low-level group of the expense.",
153 )
154 category: Optional[str] = Field(
155 default=None,
156 description="High-level group of the expense.",
157 )
158 amount: Optional[float] = Field(
159 default=None,
160 description="Amount of the expense.",
161 )
162 description: Optional[str] = Field(
163 default=None,
164 description="Description of the expense.",
165 )Sebbene in apparenza il tipo sia simile a ExpenseAdd
Oggetti di questo tipo verranno passate a CRUDHandler
# modules/crud_handler.py
201 def update(self, ID: int, data: ExpenseUpdate):
202 """Update expense selected by ID.
203
204 Parameters
205 -----------------------
206 ID : int
207 ID of the expense to update.
208 data : ExpenseUpdate
209 New data to patch the expense with. Unset fields will not
change.
210
211 Raises
212 -----------------------
213 CRUDHandlerError
214 If specified ID is not found.
215 """
216 exp = self.session.get(Expense, ID)
217 if exp is None:
218 raise CRUDHandlerError(f"ID {ID} not found")
219
220 for k, v in data.model_dump().items():
221 if v is not None:
222 setattr(exp, k, v)
223 self.session.commit()Il metodo Session.get() consente di eseguire ricerche ed estrarre valori di campi (o il modello per intero, come in questo caso) in base a una primary key (qui, l’ID).
Il risultato sarà None se la query non ha prodotto risultati, nel qual caso la funzione solleva un CRUDHandlerErrorExpenseNone (nel qual caso vengono mantenuti quelli vecchi). Le modifiche a questo punto non sono ancora applicate al database, richiedendo una chiamata finale a Expense.commit().
Delete: Rimuovere Record
L’ultima fra le operazioni CRUD implementata in CRUDHandler è la rimozione di spese esistenti, eseguita in CRUDHandler.remove() in base a un elenco di ID. Anche in questo caso dovranno essere sollevati errori nel caso in cui degli ID non siano presenti.
Ho implementato i protocolli di verifica con il requisito di annullare modifiche eseguite in una richiesta in cui non è stato trovato un ID (anche se altri sono stati trovati, e le spese associate rimosse con successo). Ciò consente di lanciare nuovamente l’intera operazione una volta corretti gli errori.
# modules/crud_handler.py
225 def remove(self, ids: list[int]):
226 """Remove selected expenses from the DB.
227
228 Parameters
229 -----------------------
230 ids : list[int]
231 IDs of the removed expenses.
232
233 Raises
234 -----------------------
235 CRUDHandlerError
236 If a specified ID is not found.
237 """
238 for ID in ids:
239 exp = self.session.get(Expense, ID)
240 if exp is None:
241 self.session.rollback()
242 raise CRUDHandlerError(f"ID {ID} not found")
243
244 self.session.delete(exp)
245
246 self.session.commit()La gestione degli errori è più complessa in questo caso. Le Expense corrispondenti agli ID passati vengono estratte e contrassegnate per l’eliminazione (l’ultimo passaggio viene eseguito chiamando il metodo Session.delete()
Ho anche implementato un metodo erase(), che rimuove tutte le spese. Questo è particolarmente utile per i test, e ho pensato che fosse meglio separarlo da remove(), per evitare cancellazioni accidentali. Come in precedenza, le istanze di Expense (in questo caso, tutte) verranno estratte e contrassegnate per l’eliminazione. Le modifiche vengono quindi confermate.
# modules/crud_handler.py
248 def erase(self):
249 """Remove all expenses from the DB and reset ID field."""
250 for exp in self.query(QueryParameters()):
251 self.session.delete(exp)
252
253 # Resetting primary key
254 self.session.execute(text("ALTER SEQUENCE expenses_id_seq
RESTART"))
255 self.session.commit()Il comando Session.execute() viene utilizzato qui per eseguire un comando SQL (racchiuso nel wrapper text() di SQLAlchemy): questo comando SQL reimposta la chiave primaria, che altrimenti continuerebbe dall’ultimo valore assegnato prima della cancellazione.
Nelle Prossime Puntate
In questo post abbiamo capito come implementare operazioni CRUD (Create, Read, Update, Delete) su un database utilizzando metodi e costrutti SQLAlchemy.
Nel prossimo post di questa serie, impareremo come scrivere un’API che accederà a queste funzioni e comunicherà con l’utente.
Autore: Adriano Angelone
Dopo aver ottenuto la Laurea Magistrale in Fisica all’Università di Pisa nel 2013, ha ricevuto il Dottorato in Fisica all’Università di Strasburgo nel 2017. Ha lavorato come ricercatore post-dottorale all’Università di Strasburgo, alla SISSA (Trieste) e all’Università Sorbona (Parigi), prima di entrare in eXact-lab come Sviluppatore di Software Scientifico nel 2023.
In eXact-lab, lavora all’ottimizzazione di codici computazionali e allo sviluppo di software di gestione dati e data engineering.
