Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/model.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import datetime | import datetime | ||||
from uuid import UUID | from uuid import UUID | ||||
from typing import Any, Dict, List, Optional, Tuple | from typing import Any, Dict, List, Optional, Tuple | ||||
import attr | import attr | ||||
import attr.converters | import attr.converters | ||||
from attrs_strict import type_validator | from attrs_strict import type_validator | ||||
@attr.s | @attr.s | ||||
class BaseSchedulerModel: | class BaseSchedulerModel: | ||||
"""Base class for database-backed objects. | |||||
These database-backed objects are defined through attrs-based attributes | |||||
that match the columns of the database 1:1. This is a (very) lightweight | |||||
ORM. | |||||
These attrs-based attributes have metadata specific to the functionality | |||||
expected from these fields in the database: | |||||
- `primary_key`: the column is a primary key; it should be filtered out | |||||
when doing an `update` of the object | |||||
- `auto_primary_key`: the column is a primary key, which is automatically handled | |||||
by the database. It will not be inserted to. This must be matched with a | |||||
database-side default value. | |||||
- `auto_now_add`: the column is a timestamp that is set to the current time when | |||||
the object is inserted, and never updated afterwards. This must be matched with | |||||
a database-side default value. | |||||
- `auto_now`: the column is a timestamp that is set to the current time when | |||||
the object is inserted or updated. | |||||
""" | |||||
_pk_cols: Optional[Tuple[str, ...]] = None | |||||
_select_cols: Optional[Tuple[str, ...]] = None | _select_cols: Optional[Tuple[str, ...]] = None | ||||
_insert_cols_and_metavars: Optional[Tuple[Tuple[str, ...], Tuple[str, ...]]] = None | _insert_cols_and_metavars: Optional[Tuple[Tuple[str, ...], Tuple[str, ...]]] = None | ||||
@classmethod | @classmethod | ||||
def primary_key_columns(cls) -> Tuple[str, ...]: | |||||
"""Get the primary key columns for this object type""" | |||||
if cls._pk_cols is None: | |||||
columns: List[str] = [] | |||||
for field in attr.fields(cls): | |||||
if any( | |||||
field.metadata.get(flag) | |||||
for flag in ("auto_primary_key", "primary_key") | |||||
): | |||||
columns.append(field.name) | |||||
cls._pk_cols = tuple(sorted(columns)) | |||||
return cls._pk_cols | |||||
@classmethod | |||||
def select_columns(cls) -> Tuple[str, ...]: | def select_columns(cls) -> Tuple[str, ...]: | ||||
"""Get all the database columns needed for a `select` on this object type""" | """Get all the database columns needed for a `select` on this object type""" | ||||
if cls._select_cols is None: | if cls._select_cols is None: | ||||
columns: List[str] = [] | columns: List[str] = [] | ||||
for field in attr.fields(cls): | for field in attr.fields(cls): | ||||
columns.append(field.name) | columns.append(field.name) | ||||
cls._select_cols = tuple(sorted(columns)) | cls._select_cols = tuple(sorted(columns)) | ||||
return cls._select_cols | return cls._select_cols | ||||
@classmethod | @classmethod | ||||
def insert_columns_and_metavars(cls) -> Tuple[Tuple[str, ...], Tuple[str, ...]]: | def insert_columns_and_metavars(cls) -> Tuple[Tuple[str, ...], Tuple[str, ...]]: | ||||
"""Get the database columns and metavars needed for an `insert` or `update` on | """Get the database columns and metavars needed for an `insert` or `update` on | ||||
this object type. | this object type. | ||||
This supports the following attributes as booleans in the field's metadata: | This implements support for the `auto_*` field metadata attributes. | ||||
- primary_key: handled by the database; never inserted or updated | |||||
- auto_now_add: handled by the database; set to now() on insertion, never | |||||
updated | |||||
- auto_now: handled by the client; set to now() on every insertion and update | |||||
""" | """ | ||||
if cls._insert_cols_and_metavars is None: | if cls._insert_cols_and_metavars is None: | ||||
zipped_cols_and_metavars: List[Tuple[str, str]] = [] | zipped_cols_and_metavars: List[Tuple[str, str]] = [] | ||||
for field in attr.fields(cls): | for field in attr.fields(cls): | ||||
if any( | if any( | ||||
field.metadata.get(flag) for flag in ("auto_now_add", "primary_key") | field.metadata.get(flag) | ||||
for flag in ("auto_now_add", "auto_primary_key") | |||||
): | ): | ||||
continue | continue | ||||
elif field.metadata.get("auto_now"): | elif field.metadata.get("auto_now"): | ||||
zipped_cols_and_metavars.append((field.name, "now()")) | zipped_cols_and_metavars.append((field.name, "now()")) | ||||
else: | else: | ||||
zipped_cols_and_metavars.append((field.name, f"%({field.name})s")) | zipped_cols_and_metavars.append((field.name, f"%({field.name})s")) | ||||
zipped_cols_and_metavars.sort() | zipped_cols_and_metavars.sort() | ||||
Show All 9 Lines | class Lister(BaseSchedulerModel): | ||||
name = attr.ib(type=str, validator=[type_validator()]) | name = attr.ib(type=str, validator=[type_validator()]) | ||||
instance_name = attr.ib(type=str, validator=[type_validator()], factory=str) | instance_name = attr.ib(type=str, validator=[type_validator()], factory=str) | ||||
# Populated by database | # Populated by database | ||||
id = attr.ib( | id = attr.ib( | ||||
type=Optional[UUID], | type=Optional[UUID], | ||||
validator=type_validator(), | validator=type_validator(), | ||||
default=None, | default=None, | ||||
metadata={"primary_key": True}, | metadata={"auto_primary_key": True}, | ||||
) | ) | ||||
current_state = attr.ib( | current_state = attr.ib( | ||||
type=Dict[str, Any], validator=[type_validator()], factory=dict | type=Dict[str, Any], validator=[type_validator()], factory=dict | ||||
) | ) | ||||
created = attr.ib( | created = attr.ib( | ||||
type=Optional[datetime.datetime], | type=Optional[datetime.datetime], | ||||
validator=[type_validator()], | validator=[type_validator()], | ||||
Show All 9 Lines |