Coverage for gib_esu/models/request_models.py: 100%
143 statements
« prev ^ index » next coverage.py v7.6.4, created at 2024-11-25 16:39 +0300
1from __future__ import annotations
3import re
4from enum import Enum
5from typing import Annotated, List, Optional, TypeVar, Union, cast
7from pydantic import (
8 AfterValidator,
9 Field,
10 PlainValidator,
11 StringConstraints,
12 model_validator,
13)
14from typing_extensions import Self
16from gib_esu.models.base_model import CustomBaseModel
18# regex patterns
20RegEx__Soket_No = r"^Soket\d+$"
21RegEx__Soket_Sayisi = r"^[1-9]$"
22RegEx__Firma_VKN = r"\b\d{10}\b"
23RegEx__Il_Kodu = r"\b\d{3}\b"
24RegEx__Tarih = r"^\d{4}-\d{2}-\d{2}$" # YYYY-MM-DD
27# validators
28def _validate_tax_number(tax_nr: str) -> str:
29 """Validates a given tax number.
31 Args:
32 tax_nr (str): Tax number to validate
34 Raises:
35 ValueError: In case tax_nr does not conform to tax number scheme
37 Returns:
38 str: Validated tax_nr
39 """
40 v_str = tax_nr.strip()
41 if not v_str or re.fullmatch(RegEx__Firma_VKN, v_str):
42 return v_str
43 raise ValueError(f"{tax_nr} geçerli bir vergi kimlik numarası değil")
46def _validate_tax_payer_and_update_models(model: CustomBaseModelWithValidator) -> None:
47 """Validates both the tax payer registration and the charge point update models."""
48 durum: Union[ESUMukellefBilgisi, ESUGuncellemeBilgisi] = (
49 cast(ESUGuncellemeModel, model).guncelleme_istek_bilgileri
50 if isinstance(model, ESUGuncellemeModel)
51 else cast(ESUMukellefModel, model).durum_bilgileri
52 )
53 sertifika_tarihi = durum.sertifika_tarihi or ""
54 sertifika_no = durum.sertifika_no or ""
55 fatura_tarihi = durum.fatura_tarihi or ""
56 fatura_ettn = durum.fatura_ettn or ""
57 mulkiyet_sahibi_vkn_tckn = durum.mulkiyet_sahibi_vkn_tckn or ""
58 mulkiyet_sahibi_ad_unvan = durum.mulkiyet_sahibi_ad_unvan or ""
60 # conditions to check to enforce consistency and mutual exclusion rules
62 mulkiyet_vkn_does_not_exist = len(mulkiyet_sahibi_vkn_tckn.strip()) == 0
63 mulkiyet_unvan_does_not_exist = len(mulkiyet_sahibi_ad_unvan.strip()) == 0
64 fatura_ettn_does_not_exist = len(fatura_ettn.strip()) == 0
65 fatura_tarihi_does_not_exist = len(fatura_tarihi.strip()) == 0
66 sertifika_tarihi_does_exist = len(sertifika_tarihi.strip()) > 0
67 sertifika_no_does_exist = len(sertifika_no.strip()) > 0
69 # enforce consistency between `fatura_ettn` and `fatura_tarihi`
70 if fatura_ettn_does_not_exist != fatura_tarihi_does_not_exist:
71 raise ValueError(
72 (
73 "`fatura_tarihi` ile `fatura_ettn` tutarsız; "
74 "ikisi de boş veya ikisi de dolu olmalı"
75 )
76 )
78 # enforce consistency between
79 # `mulkiyet_sahibi_vkn_tckn` and `mulkiyet_sahibi_ad_unvan`
80 if mulkiyet_vkn_does_not_exist != mulkiyet_unvan_does_not_exist:
81 raise ValueError(
82 (
83 "`mulkiyet_sahibi_vkn_tckn` ile `mulkiyet_sahibi_ad_unvan` tutarsız; "
84 "ikisi de boş veya ikisi de dolu olmalı"
85 )
86 )
88 # enforce consistency between `sertifika_no` and `sertifika_tarihi`
89 if sertifika_no_does_exist != sertifika_tarihi_does_exist:
90 raise ValueError(
91 "`sertifika_no` ile `sertifika_tarihi` tutarsız; "
92 "ikisi de boş veya ikisi de dolu olmalı"
93 )
95 # disallow co-existence or co-absence of
96 # `fatura_ettn` and `mulkiyet_sahibi_vkn_tckn`
97 if fatura_ettn_does_not_exist == mulkiyet_vkn_does_not_exist:
98 raise ValueError(
99 (
100 "`fatura_ettn` veya `mulkiyet_sahibi_vkn_tckn` "
101 "alanlarından biri ve yalnız biri mevcut olmalıdır"
102 )
103 )
105 # allow presence of `sertifika_no` only when `mulkiyet_sahibi_vkn_tckn` is present
106 """ if sertifika_no_does_exist and mulkiyet_vkn_does_not_exist:
107 raise ValueError(
108 "sertifika bilgilerini gönderebilmek için "
109 "`mulkiyet_sahibi_vkn_tckn` dolu olmalıdır"
110 ) """
112 # conditionally check fatura_tarihi
113 if not fatura_tarihi_does_not_exist and not bool(
114 re.match(RegEx__Tarih, fatura_tarihi)
115 ):
116 raise ValueError("`fatura_tarihi` YYYY-MM-DD formatında olmalıdır")
118 # conditionally check sertifika_tarihi
119 if sertifika_tarihi_does_exist and not bool(
120 re.match(RegEx__Tarih, sertifika_tarihi)
121 ):
122 raise ValueError("`sertifika_tarihi` YYYY-MM-DD formatında olmalıdır")
125# type definitions
127T = TypeVar("T")
129"""Type definition of a non-empty string."""
130NonEmptyString = Annotated[
131 str,
132 StringConstraints(strip_whitespace=True),
133 AfterValidator(lambda v: v),
134 StringConstraints(min_length=1),
135]
137"""Type definition of a non-empty list."""
138NonEmptyList = Annotated[List[T], Field(default_factory=list, min_length=1)]
141"""Type definition for required tax numbers."""
142TaxNumber = Annotated[
143 str,
144 StringConstraints(strip_whitespace=True),
145 Field(pattern=RegEx__Firma_VKN),
146]
149"""Type definition for optional tax numbers."""
150TaxNumberOrEmpty = Annotated[
151 str,
152 StringConstraints(strip_whitespace=True),
153 PlainValidator(_validate_tax_number),
154]
156"""Type definition for city code."""
157CityCode = Annotated[
158 str, StringConstraints(strip_whitespace=True), Field(pattern=RegEx__Il_Kodu)
159]
161# enums
164class SoketTipi(str, Enum):
165 """Socket type enum."""
167 AC = "AC"
168 DC = "DC"
171class ESUTipi(str, Enum):
172 """Charge point type enum."""
174 AC = SoketTipi.AC.value
175 DC = SoketTipi.DC.value
176 AC_DC = "AC/DC"
179# domain models
182class Soket(CustomBaseModel):
183 """Charge point's connectors."""
185 soket_no: str = Field(pattern=RegEx__Soket_No) # Soket1, Soket2, Soket3, etc.
186 soket_tip: SoketTipi
189class ESUSeriNo(CustomBaseModel):
190 """Charge point's serial number."""
192 esu_seri_no: NonEmptyString
195class ESU(ESUSeriNo):
196 """Charge point model."""
198 esu_soket_tipi: ESUTipi
199 esu_soket_sayisi: str = Field(pattern=RegEx__Soket_Sayisi) # "1", "2", "3", etc.
200 esu_soket_detay: NonEmptyList[Soket]
201 esu_markasi: NonEmptyString
202 esu_modeli: NonEmptyString
205class FirmaKodu(CustomBaseModel):
206 """Company code model."""
208 firma_kodu: NonEmptyString
211class Firma(FirmaKodu):
212 """Company info model."""
214 firma_vkn: TaxNumber
215 epdk_lisans_no: NonEmptyString
216 firma_unvan: Optional[NonEmptyString] = Field(default=None, exclude=True)
219class Lokasyon(CustomBaseModel):
220 """EV charging location model."""
222 il_kodu: CityCode
223 ilce: NonEmptyString
224 adres_numarası: Optional[str] = ""
225 koordinat: Optional[str] = ""
228class Mukellef(CustomBaseModel):
229 """Tax payer model."""
231 mukellef_vkn: Optional[TaxNumberOrEmpty] = ""
232 mukellef_unvan: Optional[str] = ""
235class Sertifika(CustomBaseModel):
236 """Certificate model."""
238 sertifika_no: Optional[str] = ""
239 sertifika_tarihi: Optional[str] = ""
242class Fatura(CustomBaseModel):
243 """Invoice model."""
245 fatura_tarihi: Optional[str] = ""
246 fatura_ettn: Optional[str] = ""
249class MulkiyetSahibi(CustomBaseModel):
250 """Charge point owner model."""
252 mulkiyet_sahibi_vkn_tckn: Optional[TaxNumberOrEmpty] = ""
253 mulkiyet_sahibi_ad_unvan: Optional[str] = ""
256class ESUMukellefBilgisi(
257 ESUSeriNo, Fatura, Lokasyon, Mukellef, MulkiyetSahibi, Sertifika
258):
259 """Intermediary model that encapsulates charge point and tax payer information."""
261 pass
264class ESUGuncellemeBilgisi(ESUSeriNo, Fatura, Lokasyon, MulkiyetSahibi, Sertifika):
265 """Intermediary model that encapsulates charge point and ownership information."""
267 pass
270class CustomBaseModelWithValidator(CustomBaseModel):
271 """Custom base model with a predefined model validator function."""
273 @model_validator(mode="after")
274 def _enforce_model_constraints(self) -> Self:
275 """Validates the model according to the model constraints."""
276 _validate_tax_payer_and_update_models(cast(CustomBaseModelWithValidator, self))
277 return self
280# request models
283class ESUKayitModel(Firma):
284 """Charge point registration request model."""
286 kayit_bilgisi: ESU
288 @classmethod
289 def olustur(cls, firma: Firma, esu: ESU) -> ESUKayitModel:
290 """Constructs a ESUKayitModel from given `esu` and `firma` arguments.
292 Args:
293 firma (Firma): Company information
294 esu (ESU): Charge point information
296 Returns:
297 ESUKayitModel: Constructed model instance
298 """
299 combined_data = {**firma.__dict__, "kayit_bilgisi": esu}
300 return ESUKayitModel(**combined_data)
302 @model_validator(mode="after")
303 def _enforce_model_constraints(self) -> Self:
304 """Validates the model according to the model constraints."""
305 kayit: ESU = self.kayit_bilgisi
306 soket_tipi = kayit.esu_soket_tipi
307 soket_sayisi = kayit.esu_soket_sayisi
308 soket_detay = kayit.esu_soket_detay
310 # check socket type integrity
311 for esu_soket in soket_detay:
312 soket_tip = esu_soket.soket_tip
313 if soket_tipi in SoketTipi.__members__.values() and soket_tip != soket_tipi:
314 raise ValueError("Soket detayları `esu_soket_tipi` ile uyumlu değil")
316 # compare socket count to socket details length
317 assert len(soket_detay) == int(
318 soket_sayisi
319 ), "`esu_soket_sayisi` kadar `esu_soket_detay` olmalı"
321 # check socket details when charge point type is AC/DC
322 if soket_tipi == ESUTipi.AC_DC.value:
323 has_ac = any(item.soket_tip == SoketTipi.AC.value for item in soket_detay)
324 has_dc = any(item.soket_tip == SoketTipi.DC.value for item in soket_detay)
325 assert has_ac and has_dc, "Soket detayları AC/DC EŞÜ ile uyumlu değil"
326 return self
329class ESUKapatmaModel(CustomBaseModel):
330 """Charge point delisting request model."""
332 firma_kodu: NonEmptyString
333 kapatma_bilgisi: ESUSeriNo
336class ESUMukellefModel(CustomBaseModelWithValidator, FirmaKodu):
337 """Charge point tax payer info registration model."""
339 durum_bilgileri: ESUMukellefBilgisi
341 @classmethod
342 def olustur(
343 cls,
344 esu_seri_no: str,
345 firma_kodu: str,
346 fatura: Fatura,
347 lokasyon: Lokasyon,
348 mukellef: Mukellef,
349 mulkiyet_sahibi: Optional[MulkiyetSahibi] = None,
350 sertifika: Optional[Sertifika] = None,
351 ) -> ESUMukellefModel:
352 """Constructs a ESUMukellefModel from given arguments.
354 Args:
355 esu_seri_no (str): Charge point serial number
356 firma_kodu (str): Company code
357 fatura (Fatura): Invoice information
358 lokasyon (Lokasyon): Location information
359 mukellef (Mukellef): Tax payer information
360 mulkiyet_sahibi (Optional[MulkiyetSahibi], optional):
361 Ownership information. Defaults to None.
362 sertifika (Optional[Sertifika], optional): Certificate. Defaults to None.
364 Returns:
365 ESUMukellefModel: Constructed model instance
366 """
367 combined_data = {
368 **ESUSeriNo(esu_seri_no=esu_seri_no).model_dump(),
369 **fatura.model_dump(),
370 **lokasyon.model_dump(),
371 **mukellef.model_dump(),
372 **(mulkiyet_sahibi or MulkiyetSahibi()).model_dump(),
373 **(sertifika or Sertifika()).model_dump(),
374 }
375 mukellef_durum = ESUMukellefBilgisi(**combined_data)
376 return ESUMukellefModel(
377 **FirmaKodu(firma_kodu=firma_kodu).model_dump(),
378 durum_bilgileri=mukellef_durum,
379 )
382class ESUGuncellemeModel(CustomBaseModelWithValidator, FirmaKodu):
383 """Charge point update request model."""
385 guncelleme_istek_bilgileri: ESUGuncellemeBilgisi
387 @classmethod
388 def olustur(
389 cls,
390 esu_seri_no: ESUSeriNo,
391 firma_kodu: str,
392 fatura: Fatura,
393 lokasyon: Lokasyon,
394 mulkiyet_sahibi: Optional[MulkiyetSahibi] = None,
395 sertifika: Optional[Sertifika] = None,
396 ) -> ESUGuncellemeModel:
397 """Constructs a ESUGuncellemeModel from given arguments.
399 Args:
400 esu_seri_no (str): Charge point serial number
401 firma_kodu (str): Company code
402 fatura (Fatura): Invoice information
403 lokasyon (Lokasyon): Location information
404 mulkiyet_sahibi (Optional[MulkiyetSahibi], optional):
405 Ownership information. Defaults to None.
406 sertifika (Optional[Sertifika], optional): Certificate. Defaults to None.
408 Returns:
409 ESUGuncellemeModel: Constructed model instance
410 """
411 combined_data = {
412 **esu_seri_no.model_dump(),
413 **fatura.model_dump(),
414 **lokasyon.model_dump(),
415 **(mulkiyet_sahibi or MulkiyetSahibi()).model_dump(),
416 **(sertifika or Sertifika()).model_dump(),
417 }
419 return ESUGuncellemeModel(
420 **FirmaKodu(firma_kodu=firma_kodu).model_dump(),
421 guncelleme_istek_bilgileri=ESUGuncellemeBilgisi(**combined_data),
422 )