Ingest comdirect #5
+57
-31
@@ -1,7 +1,7 @@
|
|||||||
from datetime import date
|
from datetime import date
|
||||||
from decimal import Decimal
|
from decimal import Decimal
|
||||||
|
|
||||||
from pydantic import Field, computed_field, field_validator
|
from pydantic import ConfigDict, Field, computed_field, field_validator
|
||||||
from sqlalchemy import Column, Date, String
|
from sqlalchemy import Column, Date, String
|
||||||
|
|
||||||
from src.models._common import (
|
from src.models._common import (
|
||||||
@@ -11,10 +11,12 @@ from src.models._common import (
|
|||||||
_parse_date_german,
|
_parse_date_german,
|
||||||
_parse_decimal_german,
|
_parse_decimal_german,
|
||||||
date2str,
|
date2str,
|
||||||
|
decimal2str,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class ComdirectBase(ExportBaseModel):
|
class ComdirectBase(ExportBaseModel):
|
||||||
|
model_config = ConfigDict(populate_by_name=True)
|
||||||
date_val: date | None = Field(
|
date_val: date | None = Field(
|
||||||
alias="Wertstellung (Valuta)", examples=["07.09.2026"]
|
alias="Wertstellung (Valuta)", examples=["07.09.2026"]
|
||||||
)
|
)
|
||||||
@@ -22,7 +24,9 @@ class ComdirectBase(ExportBaseModel):
|
|||||||
|
|
||||||
@field_validator("date_val", mode="before")
|
@field_validator("date_val", mode="before")
|
||||||
@classmethod
|
@classmethod
|
||||||
def parse_date_val(cls, date_str: str) -> date | None:
|
def parse_date_val(cls, date_str: str | None) -> date | None:
|
||||||
|
if date_str is None:
|
||||||
|
return
|
||||||
return _parse_date_german(date_str)
|
return _parse_date_german(date_str)
|
||||||
|
|
||||||
@field_validator("amount_eur", mode="before")
|
@field_validator("amount_eur", mode="before")
|
||||||
@@ -30,13 +34,28 @@ class ComdirectBase(ExportBaseModel):
|
|||||||
def parse_amount(cls, decimal_str: str) -> Decimal:
|
def parse_amount(cls, decimal_str: str) -> Decimal:
|
||||||
return _parse_decimal_german(decimal_str)
|
return _parse_decimal_german(decimal_str)
|
||||||
|
|
||||||
def parameters_to_dict(self) -> dict[str, str]: ...
|
@classmethod
|
||||||
|
def _date2str(cls, date_value: date | None) -> str:
|
||||||
|
return date2str(date_value, default="--", date_format="%d.%m.%Y")
|
||||||
|
|
||||||
def parameters_to_str(self, sep=", ", str_qualifier=False): ...
|
def parameters_to_dict(self, by_name=True) -> dict[str, str]:
|
||||||
|
return {
|
||||||
|
"date_val" if by_name else "Wertstellung (Valuta)": self._date2str(
|
||||||
|
self.date_val
|
||||||
|
),
|
||||||
|
"amount_eur" if by_name else "Umsatz in EUR": decimal2str(self.amount_eur),
|
||||||
|
}
|
||||||
|
|
||||||
def __repr__(self): ...
|
def get_orig_headers(self, sep: str = ";", bracket: str = '"'):
|
||||||
|
return sep.join(
|
||||||
|
f"{bracket}{txt}{bracket}" for txt in self.parameters_to_dict(by_name=False)
|
||||||
|
)
|
||||||
|
|
||||||
def __str__(self): ...
|
def get_orig_row(self, sep: str = ";", bracket: str = '"'):
|
||||||
|
return sep.join(
|
||||||
|
f"{bracket}{txt}{bracket}"
|
||||||
|
for txt in self.parameters_to_dict(by_name=False).values()
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class ComdirectGiro(ComdirectBase):
|
class ComdirectGiro(ComdirectBase):
|
||||||
@@ -59,21 +78,18 @@ class ComdirectGiro(ComdirectBase):
|
|||||||
def parse_date_book(cls, date_str: str) -> date | None:
|
def parse_date_book(cls, date_str: str) -> date | None:
|
||||||
return _parse_date_german(date_str)
|
return _parse_date_german(date_str)
|
||||||
|
|
||||||
def parameters_to_dict(self) -> dict[str, str]:
|
def parameters_to_dict(self, by_name=True) -> dict[str, str]:
|
||||||
return {
|
parameters = dict(**super().parameters_to_dict(by_name=by_name))
|
||||||
"Buchungstag": date2str(
|
parameters.update(
|
||||||
self.date_book, default="--", date_format="%d.%m.%Y"
|
{
|
||||||
|
"date_book" if by_name else "Buchungstag": self._date2str(
|
||||||
|
self.date_book
|
||||||
),
|
),
|
||||||
"Vorgang": self.type,
|
"type" if by_name else "Vorgang": self.type,
|
||||||
"Text": self.text,
|
"text" if by_name else "Buchungstext": self.text,
|
||||||
}
|
}
|
||||||
|
)
|
||||||
def parameters_to_str(self, sep=", ", str_qualifier=False):
|
return parameters
|
||||||
...
|
|
||||||
|
|
||||||
def __repr__(self): ...
|
|
||||||
|
|
||||||
def __str__(self): ...
|
|
||||||
|
|
||||||
|
|
||||||
class ComdirectCreditCard(ComdirectGiro):
|
class ComdirectCreditCard(ComdirectGiro):
|
||||||
@@ -96,13 +112,17 @@ class ComdirectCreditCard(ComdirectGiro):
|
|||||||
alias="Umsatz in EUR", examples=["32,43"], default=None
|
alias="Umsatz in EUR", examples=["32,43"], default=None
|
||||||
)
|
)
|
||||||
|
|
||||||
def parameters_to_dict(self) -> dict[str, str]: ...
|
|
||||||
|
|
||||||
def parameters_to_str(self, sep=", ", str_qualifier=False): ...
|
def parameters_to_str(self, sep=", ", str_qualifier=False): ...
|
||||||
|
|
||||||
def __repr__(self): ...
|
def parameters_to_dict(self) -> dict[str, str]:
|
||||||
|
parameters = dict(**super().parameters_to_dict())
|
||||||
def __str__(self): ...
|
parameters.update(
|
||||||
|
{
|
||||||
|
"Umsatztag": self._date2str(self.date_val),
|
||||||
|
"Referenz": self.reference,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return parameters
|
||||||
|
|
||||||
|
|
||||||
class ComdirectDepot(ComdirectBase):
|
class ComdirectDepot(ComdirectBase):
|
||||||
@@ -121,13 +141,19 @@ class ComdirectDepot(ComdirectBase):
|
|||||||
def parse_decimals(cls, decimal_in: str) -> Decimal:
|
def parse_decimals(cls, decimal_in: str) -> Decimal:
|
||||||
return _parse_decimal_german(decimal_in)
|
return _parse_decimal_german(decimal_in)
|
||||||
|
|
||||||
def parameters_to_dict(self) -> dict[str, str]: ...
|
def parameters_to_dict(self) -> dict[str, str]:
|
||||||
|
parameters = dict(**super().parameters_to_dict())
|
||||||
def parameters_to_str(self, sep=", ", str_qualifier=False): ...
|
parameters.update(
|
||||||
|
{
|
||||||
def __repr__(self): ...
|
"Geschäftstag": self._date2str(self.date_val),
|
||||||
|
"Stück / Nom.": decimal2str(self.num),
|
||||||
def __str__(self): ...
|
"Bezeichnung": self.name,
|
||||||
|
"WKN": self.wkn,
|
||||||
|
"Währung": self.currency,
|
||||||
|
"Ausführungskurs": decimal2str(self.spot_price),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return parameters
|
||||||
|
|
||||||
|
|
||||||
class ComdirectBaseDb(IngestdbBase):
|
class ComdirectBaseDb(IngestdbBase):
|
||||||
|
|||||||
+17
-5
@@ -1,4 +1,3 @@
|
|||||||
from collections.abc import Callable
|
|
||||||
from decimal import Decimal
|
from decimal import Decimal
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
@@ -27,6 +26,18 @@ def _amount() -> Decimal:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _amount_eur() -> str:
|
||||||
|
"""Generate German-formatted decimal string (e.g., '32,43')."""
|
||||||
|
val = B4wExportFactory.__faker__.pydecimal(
|
||||||
|
left_digits=6,
|
||||||
|
right_digits=2,
|
||||||
|
positive=True,
|
||||||
|
min_value=Decimal("0.01"),
|
||||||
|
)
|
||||||
|
# Format as German string with comma decimal separator
|
||||||
|
return f"{val:.2f}".replace(".", ",").strip()
|
||||||
|
|
||||||
|
|
||||||
def _sentence() -> str:
|
def _sentence() -> str:
|
||||||
return B4wExportFactory.__faker__.sentence(
|
return B4wExportFactory.__faker__.sentence(
|
||||||
nb_words=8,
|
nb_words=8,
|
||||||
@@ -61,16 +72,17 @@ class B4wExportFactory(ModelFactory[B4wExport]):
|
|||||||
|
|
||||||
class ComdirectGiroFactory(ModelFactory[ComdirectGiro]):
|
class ComdirectGiroFactory(ModelFactory[ComdirectGiro]):
|
||||||
__faker__ = Faker("de_DE")
|
__faker__ = Faker("de_DE")
|
||||||
__use_example__ = True
|
__use_examples__ = True
|
||||||
|
|
||||||
Buchungstext = Use(_sentence)
|
text = Use(_sentence)
|
||||||
|
amount_eur = Use(_amount_eur)
|
||||||
|
|
||||||
|
|
||||||
class ComdirectCreditCardFactory(ModelFactory[ComdirectCreditCard]):
|
class ComdirectCreditCardFactory(ModelFactory[ComdirectCreditCard]):
|
||||||
__faker__ = Faker("de_DE")
|
__faker__ = Faker("de_DE")
|
||||||
__use_example__ = True
|
__use_examples__ = True
|
||||||
|
|
||||||
|
|
||||||
class ComdirectDepotFactory(ModelFactory[ComdirectDepot]):
|
class ComdirectDepotFactory(ModelFactory[ComdirectDepot]):
|
||||||
__faker__ = Faker("de_DE")
|
__faker__ = Faker("de_DE")
|
||||||
__use_example__ = True
|
__use_examples__ = True
|
||||||
|
|||||||
@@ -0,0 +1,24 @@
|
|||||||
|
from src.models.comdirect import ComdirectGiro
|
||||||
|
from tests.factories import ComdirectGiroFactory
|
||||||
|
|
||||||
|
|
||||||
|
def test_comdirect_giro_basic_and_stringification():
|
||||||
|
data = ComdirectGiroFactory.build()
|
||||||
|
assert data
|
||||||
|
assert isinstance(data, ComdirectGiro)
|
||||||
|
orig_headers = data.get_orig_headers()
|
||||||
|
assert orig_headers
|
||||||
|
assert isinstance(orig_headers, str)
|
||||||
|
header_list = orig_headers.split(";")
|
||||||
|
assert len(header_list) > 0
|
||||||
|
assert set(header_list) == {
|
||||||
|
'"Wertstellung (Valuta)"',
|
||||||
|
'"Umsatz in EUR"',
|
||||||
|
'"Buchungstag"',
|
||||||
|
'"Vorgang"',
|
||||||
|
'"Buchungstext"',
|
||||||
|
}
|
||||||
|
orig_row = data.get_orig_row()
|
||||||
|
assert orig_row
|
||||||
|
assert isinstance(orig_row, str)
|
||||||
|
assert len(orig_row.split(";")) > 0
|
||||||
Reference in New Issue
Block a user