[Python] How to apply a working mandatory/optional fields serialization for a complex file...

I am working on a project that involves transformation of various inputs of data, transforming it into a xml file that strictly follows an xsd schema. My problem is that I want to give users a visualization of tables and fields that have many nested, and complex structures with many optional, and mandatory fields for filing out this very complex file that strictly follows an xsd schema with numerous subset of rules. I want to integrate the function of serializing these dto's as the users choose to. To have a main button for only mandatory fields serialization, or giving them the options to choose which optional fields besides the mandatory ones they want to fill in and serialize.

I am writing data transfer objects that have strict validation rules that go according to the schema to ensure validity. My issue is that I when I first wrote the dto's I was able to create a fully written out file, by inputting example data so that i can see how the dto's are serializing. I got to the point where I was able to serialize all dto's properly with their mandatory and optional fields and I achieved validity. The problems started when I tried introducing the function of serializing mandatory only fields from this complex file or mandatory plus optional fields. I don't know why i am failing. I am doing everything in Python, using the pydantic v2 library with the basexml model, importing Dict and Field as well as field and model validators. The validators are enforcing different rules that go according to the xsd schema, writing the classes with metadata that will be represented to the user. the metadata, has information about what to fill in these fields and the rules for the fields. I am also using for all dto's specific for their serialization def to xml functions that outline the serialization logic needed. Because depending on pydantic serialzation gave me a lot of trouble since it's not handling well the nested structures.

from typing import Optional, Dict
from lxml import etree
from pydantic import field_validator, model_validator, ConfigDict, Field
from pydantic_xml import BaseXmlModel, element
import logging
from src.validators.simple_types import SAFmiddle2textType, SAFlongtextType, SAFmiddle1textType, SAFshorttextType, SAFmonetaryType, SAFDateType
from decimal import Decimal
from src.data.nomenklaturi.nomenclature_manager import nomenclature_mgr

logger = logging.getLogger("saft.dto")

class AccountDTO(BaseXmlModel, tag="Account", ns="nsSAFT"):
account_id: SAFmiddle2textType = element(
tag="AccountID",
ns="nsSAFT",
metadata={
"label": "Код на аналитична сметка",
"help": "Код на аналитична счетоводна сметка, съгласно номенклатурата на НАП.",
"example": "2011",
"rules": "Задължително поле. Максимум 70 символа, цяло число, различно от 0, валиден код от NRA_Nom_Accounts.",
"options": nomenclature_mgr.get_code_name_pairs("NRA_Nom_Accounts", field="code_field")
}
)

account_description: SAFlongtextType = element(
tag="AccountDescription",
ns="nsSAFT",
metadata={
"label": "Наименование на сметката",
"help": "Наименование на сметката от счетоводната система на данъкоплатеца.",
"example": "Земя и постройки",
"rules": "Задължително поле. Максимум 255 символа."
}
)

taxpayer_account_id: SAFmiddle1textType = element(
tag="TaxpayerAccountID",
ns="nsSAFT",
metadata={
"label": "Номер на сметка",
"help": "Номер на немапирана сметка от счетоводната система на данъкоплатеца.",
"example": "201",
"rules": "Задължително поле. Максимум 35 символа, съответства на немапирана счетоводна сметка."
}
)

grouping_category: Optional[SAFmiddle1textType] = element(
tag="GroupingCategory",
ns="nsSAFT",
default=None,
metadata={
"label": "Категория за групиране",
"help": "Категория за групиране на сметката, използвана от данъкоплатеца (напр. разходни, приходни).",
"example": "Разходи",
"rules": "Незадължително поле. Максимум 35 символа."
}
)

grouping_code: Optional[SAFmiddle1textType] = element(
tag="GroupingCode",
ns="nsSAFT",
default=None,
metadata={
"label": "Код за групиране",
"help": "Конкретен код от категорията за групиране (напр. материали, резервни части).",
"example": "Материали",
"rules": "Незадължително поле. Максимум 35 символа."
}
)

account_type: SAFshorttextType = element(
tag="AccountType",
ns="nsSAFT",
metadata={
"label": "Тип на сметката",
"help": "Тип на сметката: Active (активна), Passive (пасивна) или Bifunctional (активно-пасивна).",
"example": "Bifunctional",
"rules": "Задължително поле. Валидни стойности: Active, Passive, Bifunctional.",
"options": "Active/Passive/Bifunctional"
}
)

account_creation_date: Optional[SAFDateType] = element(
tag="AccountCreationDate",
ns="nsSAFT",
default=None,
metadata={
"label": "Дата на създаване",
"help": "Дата на създаване на сметката във формат ISO 8601 (ГГГГ-ММ-ДД).",
"example": "2022-12-31",
"rules": "Незадължително поле. Формат ISO 8601 (ГГГГ-ММ-ДД)."
}
)

opening_debit_balance: Optional[SAFmonetaryType] = element(
tag="OpeningDebitBalance",
ns="nsSAFT",
default=None,
metadata={
"label": "Начално дебитно салдо",
"help": "Дебитно салдо в началото на периода в основна валута (EUR).",
"example": "1000.00",
"rules": "Незадължително поле. Положително или отрицателно със знак - десетично число с 2 цифри след десетичната точка в EUR."
}
)
opening_credit_balance: Optional[SAFmonetaryType] = element(
tag="OpeningCreditBalance",
ns="nsSAFT",
default=None,
metadata={
"label": "Начално кредитно салдо",
"help": "Кредитно салдо в началото на периода в основна валута (EUR).",
"example": "500.00",
"rules": "Незадължително поле. Положително или отрицателно със знак - десетично число с 2 цифри след десетичната точка в EUR."
}
)
closing_debit_balance: Optional[SAFmonetaryType] = element(
tag="ClosingDebitBalance",
ns="nsSAFT",
default=None,
metadata={
"label": "Крайно дебитно салдо",
"help": "Дебитно салдо в края на периода в основна валута (EUR).",
"example": "1200.00",
"rules": "Незадължително поле. Положително или отрицателно със знак - десетично число с 2 цифри след десетичната точка в EUR."
}
)
closing_credit_balance: Optional[SAFmonetaryType] = element(
tag="ClosingCreditBalance",
ns="nsSAFT",
default=None,
metadata={
"label": "Крайно кредитно салдо",
"help": "Кредитно салдо в края на периода в основна валута (EUR).",
"example": "700.00",
"rules": "Незадължително поле. Положително или отрицателно със знак - десетично число с 2 цифри след десетичната точка в EUR."
}
)

mandatory_only: Optional[bool] = Field(default=None, exclude=True)
config: Optional[Dict[str, bool]] = Field(default=None, exclude=True)

def to_xml_tree(self):
"""Custom serialization to ensure correct XML structure for AccountDTO."""
logger.debug(f"AccountDTO.to_xml_tree called, mandatory_only={self.mandatory_only}, config={self.config}")
nsmap = self.model_config.get("nsmap", {})
ns = nsmap.get("nsSAFT", "mf:nra:dgti:dxxxx:declaration:v1")
tree = etree.Element(f"{{{ns}}}Account", nsmap=nsmap)
config = self.config or {}

# Follow XSD order: AccountID, AccountDescription, TaxpayerAccountID, GroupingCategory, GroupingCode, AccountType, AccountCreationDate, OpeningDebitBalance, OpeningCreditBalance, ClosingDebitBalance, ClosingCreditBalance
# AccountID (mandatory)
etree.SubElement(tree, f"{{{ns}}}AccountID").text = self.account_id
logger.debug(f"Added AccountID: {self.account_id}")

# AccountDescription (mandatory)
etree.SubElement(tree, f"{{{ns}}}AccountDescription").text = self.account_description
logger.debug(f"Added AccountDescription: {self.account_description}")

# TaxpayerAccountID (mandatory)
etree.SubElement(tree, f"{{{ns}}}TaxpayerAccountID").text = self.taxpayer_account_id
logger.debug(f"Added TaxpayerAccountID: {self.taxpayer_account_id}")

# GroupingCategory (optional)
if not self.mandatory_only and config.get("include_grouping_category", True):
if self.grouping_category:
etree.SubElement(tree, f"{{{ns}}}GroupingCategory").text = self.grouping_category
logger.debug(f"Added GroupingCategory: {self.grouping_category}")
else:
etree.SubElement(tree, f"{{{ns}}}GroupingCategory")
logger.debug("Added empty GroupingCategory tag")

# GroupingCode (optional)
if not self.mandatory_only and config.get("include_grouping_code", True):
if self.grouping_code:
etree.SubElement(tree, f"{{{ns}}}GroupingCode").text = self.grouping_code
logger.debug(f"Added GroupingCode: {self.grouping_code}")
else:
etree.SubElement(tree, f"{{{ns}}}GroupingCode")
logger.debug("Added empty GroupingCode tag")

# AccountType (mandatory)
etree.SubElement(tree, f"{{{ns}}}AccountType").text = self.account_type
logger.debug(f"Added AccountType: {self.account_type}")

# AccountCreationDate (optional)
if not self.mandatory_only and config.get("include_account_creation_date", True):
if self.account_creation_date:
etree.SubElement(tree, f"{{{ns}}}AccountCreationDate").text = self.account_creation_date.isoformat()
logger.debug(f"Added AccountCreationDate: {self.account_creation_date}")
else:
etree.SubElement(tree, f"{{{ns}}}AccountCreationDate")
logger.debug("Added empty AccountCreationDate tag")

# OpeningDebitBalance (optional)
if not self.mandatory_only and config.get("include_opening_debit_balance", True):
if self.opening_debit_balance is not None:
etree.SubElement(tree, f"{{{ns}}}OpeningDebitBalance").text = f"{self.opening_debit_balance:.2f}"
logger.debug(f"Added OpeningDebitBalance: {self.opening_debit_balance}")
else:
etree.SubElement(tree, f"{{{ns}}}OpeningDebitBalance")
logger.debug("Added empty OpeningDebitBalance tag")

# OpeningCreditBalance (optional)
if not self.mandatory_only and config.get("include_opening_credit_balance", True):
if self.opening_credit_balance is not None:
etree.SubElement(tree, f"{{{ns}}}OpeningCreditBalance").text = f"{self.opening_credit_balance:.2f}"
logger.debug(f"Added OpeningCreditBalance: {self.opening_credit_balance}")
else:
etree.SubElement(tree, f"{{{ns}}}OpeningCreditBalance")
logger.debug("Added empty OpeningCreditBalance tag")

# ClosingDebitBalance (optional)
if not self.mandatory_only and config.get("include_closing_debit_balance", True):
if self.closing_debit_balance is not None:
etree.SubElement(tree, f"{{{ns}}}ClosingDebitBalance").text = f"{self.closing_debit_balance:.2f}"
logger.debug(f"Added ClosingDebitBalance: {self.closing_debit_balance}")
else:
etree.SubElement(tree, f"{{{ns}}}ClosingDebitBalance")
logger.debug("Added empty ClosingDebitBalance tag")

# ClosingCreditBalance (optional)
if not self.mandatory_only and config.get("include_closing_credit_balance", True):
if self.closing_credit_balance is not None:
etree.SubElement(tree, f"{{{ns}}}ClosingCreditBalance").text = f"{self.closing_credit_balance:.2f}"
logger.debug(f"Added ClosingCreditBalance: {self.closing_credit_balance}")
else:
etree.SubElement(tree, f"{{{ns}}}ClosingCreditBalance")
logger.debug("Added empty ClosingCreditBalance tag")

logger.debug(f"AccountDTO.to_xml_tree output: {etree.tostring(tree, encoding='unicode', pretty_print=True)}")
return tree


@field_validator("account_id")
@classmethod
def validate_account_id(cls, v, info):
"""Проверява валидността на AccountID."""
if not v or not v.strip():
logger.error(f"❌ {info.field_name} не може да бъде празен.")
raise ValueError(f"{info.field_name} не може да бъде празен.")
if not v.strip().isdigit() or int(v.strip()) == 0:
logger.error(f"❌ {info.field_name} трябва да бъде цяло число, различно от 0.")
raise ValueError(f"{info.field_name} трябва да бъде цяло число, различно от 0.")
if not nomenclature_mgr.is_valid_code("NRA_Nom_Accounts", v.strip()):
logger.error(f"❌ {info.field_name} трябва да бъде валиден код от NRA_Nom_Accounts: {v}")
raise ValueError(f"{info.field_name} трябва да бъде валиден код от NRA_Nom_Accounts.")
logger.info(f"✅ Валидиран {info.field_name}: {v}")
return v.strip()

@field_validator("account_description")
@classmethod
def validate_account_description(cls, v, info):
"""Проверява валидността на AccountDescription."""
if not v or not v.strip():
logger.error(f"❌ {info.field_name} не може да бъде празен.")
raise ValueError(f"{info.field_name} не може да бъде празен.")
logger.info(f"✅ Валидиран {info.field_name}: {v}")
return v.strip()

@field_validator("taxpayer_account_id")
@classmethod
def validate_taxpayer_account_id(cls, v, info):
"""Проверява валидността на TaxpayerAccountID."""
if not v or not v.strip():
logger.error(f"❌ {info.field_name} не може да бъде празен.")
raise ValueError(f"{info.field_name} не може да бъде празен.")
logger.info(f"✅ Валидиран {info.field_name}: {v}")
return v.strip()

@field_validator("grouping_category")
@classmethod
def validate_grouping_category(cls, v, info):
"""Проверява валидността на GroupingCategory."""
if v and not v.strip():
logger.error(f"❌ {info.field_name} не може да бъде празен, ако е подаден.")
raise ValueError(f"{info.field_name} не може да бъде празен.")
logger.info(f"✅ Валидиран {info.field_name}: {v}")
return v.strip() if v else v

@field_validator("grouping_code")
@classmethod
def validate_grouping_code(cls, v, info):
"""Проверява валидността на GroupingCode."""
if v and not v.strip():
logger.error(f"❌ {info.field_name} не може да бъде празен, ако е подаден.")
raise ValueError(f"{info.field_name} не може да бъде празен.")
logger.info(f"✅ Валидиран {info.field_name}: {v}")
return v.strip() if v else v

@field_validator("account_type")
@classmethod
def validate_account_type(cls, v, info):
"""Проверява валидността на AccountType."""
valid_options = {"Active", "Passive", "Bifunctional"}
if v not in valid_options:
logger.error(f"❌ {info.field_name} трябва да бъде един от: {valid_options}")
raise ValueError(f"{info.field_name} трябва да бъде един от: {valid_options}")
logger.info(f"✅ Валидиран {info.field_name}: {v}")
return v

@field_validator("account_creation_date")
@classmethod
def validate_account_creation_date(cls, v, info):
"""Проверява валидността на AccountCreationDate."""
if v is not None:
logger.info(f"✅ Валидиран {info.field_name}: {v}")
return v

@field_validator("opening_debit_balance", "opening_credit_balance", "closing_debit_balance", "closing_credit_balance")
@classmethod
def validate_balance(cls, v, info):
"""Проверява валидността на балансите."""
if v is not None:
if not isinstance(v, (int, float, Decimal)):
logger.error(f"❌ {info.field_name} трябва да бъде десетично число.")
raise ValueError(f"{info.field_name} трябва да бъде десетично число.")
if v < 0:
logger.error(f"❌ {info.field_name} трябва да бъде положително число.")
raise ValueError(f"{info.field_name} трябва да бъде положително число.")
logger.info(f"✅ Валидиран {info.field_name}: {float(v):.2f}")
return round(v, 2)
return v

@model_validator(mode="after")
def validate_balances_exclusive(self):
"""Проверява, че само едно от OpeningDebitBalance/OpeningCreditBalance и ClosingDebitBalance/ClosingCreditBalance е ненулево."""
opening_debit = self.opening_debit_balance
opening_credit = self.opening_credit_balance
opening_debit_is_set = opening_debit is not None and opening_debit != Decimal('0')
opening_credit_is_set = opening_credit is not None and opening_credit != Decimal('0')
if not opening_debit_is_set and not opening_credit_is_set:
logger.error("❌ Трябва да има поне OpeningDebitBalance или OpeningCreditBalance с ненулева стойност.")
raise ValueError("❌ Трябва да има поне OpeningDebitBalance или OpeningCreditBalance с ненулева стойност.")
if opening_debit_is_set and opening_credit_is_set:
logger.error("❌ Не може да има едновременно ненулеви OpeningDebitBalance и OpeningCreditBalance.")
raise ValueError("❌ Не може да има едновременно ненулеви OpeningDebitBalance и OpeningCreditBalance.")

closing_debit = self.closing_debit_balance
closing_credit = self.closing_credit_balance
closing_debit_is_set = closing_debit is not None and closing_debit != Decimal('0')
closing_credit_is_set = closing_credit is not None and closing_credit != Decimal('0')
if not closing_debit_is_set and not closing_credit_is_set:
logger.error("❌ Трябва да има поне ClosingDebitBalance или ClosingCreditBalance с ненулева стойност.")
raise ValueError("❌ Трябва да има поне ClosingDebitBalance или ClosingCreditBalance с ненулева стойност.")
if closing_debit_is_set and closing_credit_is_set:
logger.error("❌ Не може да има едновременно ненулеви ClosingDebitBalance и ClosingCreditBalance.")
raise ValueError("❌ Не може да има едновременно ненулеви ClosingDebitBalance и ClosingCreditBalance.")

logger.info("✅ Успешно валидирани начални и крайни салда.")
return self

model_config = ConfigDict(
nsmap={"nsSAFT": "mf:nra:dgti:dxxxx:declaration:v1"},
skip_empty=True,
arbitrary_types_allowed=True,
)


In these functions i outline the mandatory and optional fields and i have config list that when mandatory_only=False it has to serialize everything and when it's True it has to serialize only mandatory fields. I successfuly generate a mandatory only fields file now, but when putting it to False it messes up the sequence of the elements leading to an invalid file and i am almost sure that I wrote the def to xml functions properly respecting the needed order for validity. All of this is making me wonder how are services like mine built? How to ensure validity throughout the serialization, regardless of whether mandatory only are serialized or optional as well? How do you do this?

I have been coding for 4 months, I am very new to everything. If anyone thinks of anything that can help me, please share.

Here's a dto for refference, so you can see what I am actually writing. I am bulgarian so the probably the text won't mean anything to you. How are services that give optional and mandatory functions to the user written. I am still working on the Python service, so I want to know what is the way of introducing this function that can later be visualized through an UI and ensure validity of serialization regardless of whether they choose mandatory only or optional choices? What should I do?

Continue reading...