1
1
"""QualiCharge schemas utilities."""
2
2
3
3
import logging
4
- from typing import Generator , List , NamedTuple , Optional , Set , Tuple , Type
4
+ from typing import Generator , List , NamedTuple , Optional , Set , Tuple , Type , cast
5
5
6
6
from sqlalchemy import func
7
7
from sqlalchemy .exc import MultipleResultsFound
8
+ from sqlalchemy .schema import Column as SAColumn
8
9
from sqlmodel import Session , SQLModel , select
9
10
10
11
from ..exceptions import (
@@ -153,7 +154,7 @@ def save_statique(session: Session, statique: Statique) -> Statique:
153
154
session .commit ()
154
155
session .refresh (pdc )
155
156
156
- return pdc_to_statique (pdc )
157
+ return pdc_to_statique (cast ( PointDeCharge , pdc ) )
157
158
158
159
159
160
def update_statique (
@@ -169,7 +170,7 @@ def update_statique(
169
170
# Check that the statique to update exists
170
171
if (
171
172
session .exec (
172
- select (func .count (PointDeCharge .id )).where (
173
+ select (func .count (cast ( SAColumn , PointDeCharge .id ) )).where (
173
174
PointDeCharge .id_pdc_itinerance == id_pdc_itinerance
174
175
)
175
176
).one ()
@@ -192,7 +193,7 @@ def save_statiques(
192
193
# Ignore already existing PDC
193
194
db_pdcs = session .exec (
194
195
select (PointDeCharge .id_pdc_itinerance ).filter (
195
- PointDeCharge .id_pdc_itinerance .in_ (submitted_pdcs )
196
+ cast ( SAColumn , PointDeCharge .id_pdc_itinerance ) .in_ (submitted_pdcs )
196
197
)
197
198
).all ()
198
199
statiques = list (filter (lambda s : s .id_pdc_itinerance not in db_pdcs , statiques ))
@@ -232,7 +233,10 @@ class StatiqueSchemasEntryIndex(NamedTuple):
232
233
enseigne = Enseigne (** statique .get_fields_for_schema (Enseigne ))
233
234
localisation = Localisation (** statique .get_fields_for_schema (Localisation ))
234
235
235
- indexes = ()
236
+ indexes = []
237
+ # FIXME
238
+ # Looks like mypy does not recognize types for tuple of tuples to unpack,
239
+ # ignored unrelevant typing errors.
236
240
for entry , entries in (
237
241
(pdc , points_de_charge ),
238
242
(station , stations ),
@@ -241,12 +245,16 @@ class StatiqueSchemasEntryIndex(NamedTuple):
241
245
(enseigne , enseignes ),
242
246
(localisation , localisations ),
243
247
):
244
- if entry not in entries :
245
- entries .append (entry )
246
- indexes += (entries .index (entry ),)
248
+ if entry not in entries : # type: ignore[operator]
249
+ entries .append (entry ) # type: ignore[attr-defined]
250
+ indexes . append (entries .index (entry )) # type: ignore[attr-defined]
247
251
statiques_db_refs .append (StatiqueSchemasEntryIndex (* indexes ))
248
252
249
253
# Create database entries for each schema
254
+ #
255
+ # FIXME
256
+ # Looks like mypy does not recognize types for tuple of tuples to unpack,
257
+ # ignored unrelevant typing errors.
250
258
for entries , fields in (
251
259
(points_de_charge , {"id_pdc_itinerance" }),
252
260
(stations , {"id_station_itinerance" }),
@@ -255,12 +263,12 @@ class StatiqueSchemasEntryIndex(NamedTuple):
255
263
(enseignes , None ),
256
264
(localisations , {"adresse_station" }),
257
265
):
258
- for idx , entry in enumerate (entries ):
266
+ for idx , entry in enumerate (entries ): # type: ignore[arg-type]
259
267
_ , db_entry = get_or_create (
260
268
session , entry , fields , add = False , commit = False , refresh = False
261
269
)
262
- entries [idx ] = db_entry
263
- session .add_all (entries )
270
+ entries [idx ] = db_entry # type: ignore[index]
271
+ session .add_all (entries ) # type: ignore[arg-type]
264
272
265
273
# Handle relationships
266
274
for (
0 commit comments