Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/model.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020-2021 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 typing import Any, Dict, List, Optional, Tuple, Union | from typing import Any, Dict, List, Optional, Tuple, Union | ||||
from uuid import UUID | from uuid import UUID | ||||
import attr | import attr | ||||
import attr.converters | import attr.converters | ||||
from attrs_strict import type_validator | from attrs_strict import type_validator | ||||
def check_timestamptz(value) -> None: | |||||
"""Checks the date has a timezone.""" | |||||
if value is not None and value.tzinfo is None: | |||||
raise ValueError("date must be a timezone-aware datetime.") | |||||
@attr.s | @attr.s | ||||
class BaseSchedulerModel: | class BaseSchedulerModel: | ||||
"""Base class for database-backed objects. | """Base class for database-backed objects. | ||||
These database-backed objects are defined through attrs-based attributes | These database-backed objects are defined through attrs-based attributes | ||||
that match the columns of the database 1:1. This is a (very) lightweight | that match the columns of the database 1:1. This is a (very) lightweight | ||||
ORM. | ORM. | ||||
▲ Show 20 Lines • Show All 167 Lines • ▼ Show 20 Lines | class PaginatedListedOriginList(BaseSchedulerModel): | ||||
origins = attr.ib(type=List[ListedOrigin], validator=[type_validator()]) | origins = attr.ib(type=List[ListedOrigin], validator=[type_validator()]) | ||||
next_page_token = attr.ib( | next_page_token = attr.ib( | ||||
type=Optional[ListedOriginPageToken], | type=Optional[ListedOriginPageToken], | ||||
validator=[type_validator()], | validator=[type_validator()], | ||||
converter=convert_listed_origin_page_token, | converter=convert_listed_origin_page_token, | ||||
default=None, | default=None, | ||||
) | ) | ||||
@attr.s(frozen=True, slots=True) | |||||
class OriginVisitStats(BaseSchedulerModel): | |||||
"""Represents an aggregated origin visits view. | |||||
""" | |||||
url = attr.ib( | |||||
type=str, validator=[type_validator()], metadata={"primary_key": True} | |||||
) | |||||
visit_type = attr.ib( | |||||
type=str, validator=[type_validator()], metadata={"primary_key": True} | |||||
) | |||||
last_eventful = attr.ib( | |||||
type=Optional[datetime.datetime], validator=type_validator() | |||||
) | |||||
last_uneventful = attr.ib( | |||||
type=Optional[datetime.datetime], validator=type_validator() | |||||
) | |||||
last_failed = attr.ib(type=Optional[datetime.datetime], validator=type_validator()) | |||||
@last_eventful.validator | |||||
def check_last_eventful(self, attribute, value): | |||||
check_timestamptz(value) | |||||
@last_uneventful.validator | |||||
def check_last_uneventful(self, attribute, value): | |||||
check_timestamptz(value) | |||||
@last_failed.validator | |||||
def check_last_failed(self, attribute, value): | |||||
check_timestamptz(value) |