diff --git a/pytm/flows.py b/pytm/flows.py index b05c0ca..9ef3d85 100644 --- a/pytm/flows.py +++ b/pytm/flows.py @@ -2,7 +2,7 @@ from pytm import Element -def req_reply(src: Element, dest: Element, req_name: str, reply_name=None) -> (DF, DF): +def req_reply(src: Element, dest: Element, req_name: str, reply_name=None) -> tuple[DF, DF]: """ This function creates two datflows where one dataflow is a request and the second dataflow is the corresponding reply to the newly created request. @@ -31,7 +31,7 @@ def req_reply(src: Element, dest: Element, req_name: str, reply_name=None) -> (D return req, reply -def reply(req: DF, **kwargs) -> DF: +def reply(req: DF, **kwargs) -> tuple[DF, DF]: """ This function takes a dataflow as an argument and returns a new dataflow, which is a response to the given dataflow. diff --git a/pytm/pytm.py b/pytm/pytm.py index d710c8c..0b08080 100644 --- a/pytm/pytm.py +++ b/pytm/pytm.py @@ -9,7 +9,7 @@ from typing import ClassVar from pydantic import ValidationError -from pydantic.fields import PydanticUndefined +from pydantic_core import PydanticUndefined from collections import defaultdict from collections.abc import Iterable, Mapping diff --git a/pytm/template_engine.py b/pytm/template_engine.py index 172d152..da2c6d6 100644 --- a/pytm/template_engine.py +++ b/pytm/template_engine.py @@ -14,25 +14,25 @@ class SuperFormatter(string.Formatter): """Lightweight formatter with helpers for reports and templates.""" def format_field( - self, value: Any, spec: str + self, value: Any, format_spec: str ) -> Any: # noqa: D401 - same semantics as base - if not spec: - return super().format_field(value, spec) + if not format_spec: + return super().format_field(value, format_spec) - if spec.startswith("repeat"): - return self._format_repeat(value, spec) + if format_spec.startswith("repeat"): + return self._format_repeat(value, format_spec) - if spec.startswith("call:"): - return self._format_call(value, spec) + if format_spec.startswith("call:"): + return self._format_call(value, format_spec) - if spec.startswith("if") or spec.startswith("not"): - return self._format_conditional(value, spec) + if format_spec.startswith("if") or format_spec.startswith("not"): + return self._format_conditional(value, format_spec) - return super().format_field(value, spec) + return super().format_field(value, format_spec) - def _format_repeat(self, value: Any, spec: str) -> str: + def _format_repeat(self, value: Any, format_spec: str) -> str: """Handle the custom repeat operator.""" - template = spec.partition(":")[2] + template = format_spec.partition(":")[2] if isinstance(value, dict): iterable: Iterable[Any] = value.items() elif isinstance(value, Iterable) and not isinstance(value, (str, bytes)): @@ -41,9 +41,9 @@ def _format_repeat(self, value: Any, spec: str) -> str: iterable = [] return "".join(self.format(template, item=item) for item in iterable) - def _format_call(self, value: Any, spec: str) -> Any: + def _format_call(self, value: Any, format_spec: str) -> Any: """Evaluate callable values or report utility helpers.""" - _, _, remainder = spec.partition(":") + _, _, remainder = format_spec.partition(":") if callable(value): result = value() @@ -56,11 +56,11 @@ def _format_call(self, value: Any, spec: str) -> Any: return "".join(self.format(template, item=item) for item in result) return result - def _format_conditional(self, value: Any, spec: str) -> str: + def _format_conditional(self, value: Any, format_spec: str) -> str: """Render content conditionally based on truthiness of *value*.""" - _, _, template = spec.partition(":") + _, _, template = format_spec.partition(":") result = value() if callable(value) else value - if spec.startswith("if"): + if format_spec.startswith("if"): return template if result else "" return template if not result else "" diff --git a/pytm/tm.py b/pytm/tm.py index fb75e0c..703bad8 100644 --- a/pytm/tm.py +++ b/pytm/tm.py @@ -105,6 +105,7 @@ class TM(BaseModel, metaclass=TMModelMetaclass): _state: ClassVar[TMState] = TMState() _state_attributes: ClassVar[Dict[str, _StateAttribute]] = {} + _duplicate_ignored_attrs: ClassVar[tuple[str, ...]] = () @classmethod def _register_state_attribute(cls, name: str, descriptor: _StateAttribute) -> None: