starred/WYGIWYH
1
0
Fork 0
mirror of https://github.com/eitchtee/WYGIWYH.git synced 2026-06-08 23:52:51 +02:00
Code Issues 14 Packages Projects Releases 10 Wiki Activity

feat(import:v1:service): respect create and type fields

Browse Source
This commit is contained in:
Herculino Trotta
2025-01-23 22:48:23 -03:00
parent ba2d654f15
commit 1dc03b0a84
1 changed files with 122 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/apps/import_app/services/v1.py
+122 -37
View File
@@ -23,6 +23,12 @@ from apps.transactions.models import (
TransactionEntity, TransactionEntity,
) )
from apps.rules.signals import transaction_created from apps.rules.signals import transaction_created
from apps.import_app.schemas.v1 import (
TransactionCategoryMapping,
TransactionAccountMapping,
TransactionTagsMapping,
TransactionEntitiesMapping,
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -184,40 +190,127 @@ class ImportService:
entities = [] entities = []
# Handle related objects first # Handle related objects first
if "category" in data: if "category" in data:
category_name = data.pop("category") if "category" in data:
category, _ = TransactionCategory.objects.get_or_create(name=category_name) category_name = data.pop("category")
data["category"] = category category_mapping = next(
self.import_run.categories.add(category) (
m
for m in self.mapping.values()
if isinstance(m, TransactionCategoryMapping)
and m.target == "category"
),
None,
)
print(category_mapping)
try:
if category_mapping:
if category_mapping.type == "id":
category = TransactionCategory.objects.get(id=category_name)
else: # name
if getattr(category_mapping, "create", False):
category, _ = TransactionCategory.objects.get_or_create(
name=category_name
)
else:
category = TransactionCategory.objects.filter(
name=category_name
).first()
if category:
data["category"] = category
self.import_run.categories.add(category)
except (TransactionCategory.DoesNotExist, ValueError):
# Ignore if category doesn't exist and create is False or not set
data["category"] = None
if "account" in data: if "account" in data:
account_id = data.pop("account") account_id = data.pop("account")
account = None account_mapping = next(
if isinstance(account_id, str): (
account = Account.objects.get(name=account_id) m
elif isinstance(account_id, int): for m in self.mapping.values()
account = Account.objects.get(id=account_id) if isinstance(m, TransactionAccountMapping)
data["account"] = account and m.target == "account"
# self.import_run.acc.add(category) ),
None,
)
try:
if account_mapping and account_mapping.type == "id":
account = Account.objects.filter(id=account_id).first()
else: # name
account = Account.objects.filter(name=account_id).first()
if account:
data["account"] = account
except ValueError:
# Ignore if account doesn't exist
pass
if "tags" in data: if "tags" in data:
tag_names = data.pop("tags") tag_names = data.pop("tags")
tags_mapping = next(
(
m
for m in self.mapping.values()
if isinstance(m, TransactionTagsMapping) and m.target == "tags"
),
None,
)
for tag_name in tag_names: for tag_name in tag_names:
tag, _ = TransactionTag.objects.get_or_create(name=tag_name.strip()) try:
tags.append(tag) if tags_mapping:
self.import_run.tags.add(tag) if tags_mapping.type == "id":
tag = TransactionTag.objects.filter(id=tag_name).first()
else: # name
if getattr(tags_mapping, "create", False):
tag, _ = TransactionTag.objects.get_or_create(
name=tag_name.strip()
)
else:
tag = TransactionTag.objects.filter(
name=tag_name.strip()
).first()
if tag:
tags.append(tag)
self.import_run.tags.add(tag)
except ValueError:
# Ignore if tag doesn't exist and create is False or not set
continue
if "entities" in data: if "entities" in data:
entity_names = data.pop("entities") entity_names = data.pop("entities")
for entity_name in entity_names: entities_mapping = next(
entity, _ = TransactionEntity.objects.get_or_create( (
name=entity_name.strip() m
) for m in self.mapping.values()
entities.append(entity) if isinstance(m, TransactionEntitiesMapping)
self.import_run.entities.add(entity) and m.target == "entities"
),
None,
)
if "amount" in data: for entity_name in entity_names:
amount = data.pop("amount") try:
data["amount"] = abs(Decimal(amount)) if entities_mapping:
if getattr(entities_mapping, "create", False):
entity, _ = TransactionEntity.objects.get_or_create(
name=entity_name.strip()
)
else:
entity = TransactionEntity.objects.filter(
name=entity_name.strip()
).first()
if entity:
entities.append(entity)
self.import_run.entities.add(entity)
except ValueError:
# Ignore if entity doesn't exist and create is False or not set
continue
# Create the transaction # Create the transaction
new_transaction = Transaction.objects.create(**data) new_transaction = Transaction.objects.create(**data)
@@ -308,31 +401,23 @@ class ImportService:
coerce_to = mapping.coerce_to coerce_to = mapping.coerce_to
if "|" in coerce_to: return self._coerce_single_type(value, coerce_to, mapping)
types = coerce_to.split("|")
for t in types:
try:
return self._coerce_single_type(value, t, mapping)
except ValueError:
continue
raise ValueError(
f"Could not coerce '{value}' to any of the types: {coerce_to}"
)
else:
return self._coerce_single_type(value, coerce_to, mapping)
@staticmethod
def _coerce_single_type( def _coerce_single_type(
self, value: str, coerce_to: str, mapping: version_1.ColumnMapping value: str, coerce_to: str, mapping: version_1.ColumnMapping
) -> Union[str, int, bool, Decimal, datetime.date, list]: ) -> Union[str, int, bool, Decimal, datetime.date, list]:
if coerce_to == "str": if coerce_to == "str":
return str(value) return str(value)
elif coerce_to == "int": elif coerce_to == "int":
return int(value)
elif coerce_to == "str|int":
if hasattr(mapping, "type") and mapping.type == "id": if hasattr(mapping, "type") and mapping.type == "id":
return int(value) return int(value)
elif hasattr(mapping, "type") and mapping.type in ["name", "code"]: elif hasattr(mapping, "type") and mapping.type in ["name", "code"]:
return str(value) return str(value)
else: else:
return int(value) return str(value)
elif coerce_to == "bool": elif coerce_to == "bool":
return value.lower() in ["true", "1", "yes", "y", "on"] return value.lower() in ["true", "1", "yes", "y", "on"]
elif coerce_to == "positive_decimal": elif coerce_to == "positive_decimal":