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 | ||||
from typing import List, Optional, Tuple | import datetime | ||||
from uuid import UUID | |||||
from typing import Any, Dict, List, Optional, Tuple | |||||
import attr | import attr | ||||
import attr.converters | import attr.converters | ||||
from attrs_strict import type_validator | |||||
@attr.s | @attr.s | ||||
class BaseSchedulerModel: | class BaseSchedulerModel: | ||||
_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 | ||||
Show All 32 Lines | def insert_columns_and_metavars(cls) -> Tuple[Tuple[str, ...], Tuple[str, ...]]: | ||||
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() | ||||
cols, metavars = zip(*zipped_cols_and_metavars) | cols, metavars = zip(*zipped_cols_and_metavars) | ||||
cls._insert_cols_and_metavars = cols, metavars | cls._insert_cols_and_metavars = cols, metavars | ||||
return cls._insert_cols_and_metavars | return cls._insert_cols_and_metavars | ||||
@attr.s | |||||
class Lister(BaseSchedulerModel): | |||||
name = attr.ib(type=str, validator=[type_validator()]) | |||||
arguments = attr.ib(type=Dict[str, str], validator=[type_validator()], factory=dict) | |||||
# Populated by database | |||||
id = attr.ib( | |||||
type=Optional[UUID], | |||||
validator=type_validator(), | |||||
default=None, | |||||
metadata={"primary_key": True}, | |||||
) | |||||
current_state = attr.ib( | |||||
type=Dict[str, Any], validator=[type_validator()], factory=dict | |||||
) | |||||
created = attr.ib( | |||||
type=Optional[datetime.datetime], | |||||
validator=[type_validator()], | |||||
default=None, | |||||
metadata={"auto_now_add": True}, | |||||
) | |||||
updated = attr.ib( | |||||
type=Optional[datetime.datetime], | |||||
validator=[type_validator()], | |||||
default=None, | |||||
metadata={"auto_now": True}, | |||||
) | |||||
@name.validator | |||||
def check_name(self, attribute, value): | |||||
if not value: | |||||
raise ValueError("Lister name cannot be empty") |