getattr(condition, name).history_from_row(row_)
return condition
- @classmethod
- def all(cls, db_conn: DatabaseConnection) -> list[Condition]:
- """Collect all Conditions and their VersionedAttributes."""
- conditions = {}
- for id_, condition in cls.cache_.items():
- conditions[id_] = condition
- already_recorded = conditions.keys()
- for id_ in db_conn.column_all('conditions', 'id'):
- if id_ not in already_recorded:
- condition = cls.by_id(db_conn, id_)
- assert isinstance(condition.id_, int)
- conditions[condition.id_] = condition
- return list(conditions.values())
-
@classmethod
def by_id(cls, db_conn: DatabaseConnection, id_: int | None,
create: bool = False) -> Condition:
"""Individual days defined by their dates."""
table_name = 'days'
to_save = ['comment']
- id_type = str
def __init__(self, date: str, comment: str = '') -> None:
self.id_: str = valid_date(date)
def empty_cache(cls) -> None:
"""Empty class's cache."""
cls.cache_ = {}
+
+ @classmethod
+ def all(cls: type[BaseModelInstance],
+ db_conn: DatabaseConnection) -> list[BaseModelInstance]:
+ """Collect all objects of class."""
+ items: dict[BaseModelId, BaseModelInstance] = {}
+ for k, v in cls.cache_.items():
+ assert isinstance(v, cls)
+ items[k] = v
+ already_recorded = items.keys()
+ for id_ in db_conn.column_all(cls.table_name, 'id'):
+ if id_ not in already_recorded:
+ # pylint: disable=no-member
+ item = cls.by_id(db_conn, id_) # type: ignore[attr-defined]
+ items[item.id_] = item
+ return list(items.values())
self.enables: list[Condition] = []
self.disables: list[Condition] = []
- @classmethod
- def all(cls, db_conn: DatabaseConnection) -> list[Process]:
- """Collect all Processes and their connected VersionedAttributes."""
- processes = {}
- for id_, process in cls.cache_.items():
- processes[id_] = process
- already_recorded = processes.keys()
- for id_ in db_conn.column_all('processes', 'id'):
- if id_ not in already_recorded:
- process = cls.by_id(db_conn, id_)
- assert isinstance(process.id_, int)
- processes[process.id_] = process
- return list(processes.values())
-
@classmethod
def by_id(cls, db_conn: DatabaseConnection, id_: int | None,
create: bool = False) -> Process: