| |
@@ -1,20 +1,22 @@
|
| |
import abc
|
| |
- from datetime import datetime
|
| |
from dataclasses import dataclass
|
| |
+ from datetime import datetime
|
| |
from enum import Enum
|
| |
- from typing import Tuple, Union
|
| |
+ from typing import Tuple, Union, NoReturn
|
| |
from urllib.parse import urljoin
|
| |
|
| |
import pandas as pd
|
| |
import requests
|
| |
|
| |
- from contributor_trends.utils.exceptions import InvalidDatasetFormat
|
| |
+ from contributor_trends.dashboard.utils.exceptions import InvalidDatasetFormat
|
| |
+
|
| |
+ __all__ = ["Dataset", "ActiveContributors"]
|
| |
|
| |
|
| |
@dataclass
|
| |
class DatasetMeta:
|
| |
filepath: str
|
| |
- updated_at: datetime
|
| |
+ updated_at: Union[datetime, str]
|
| |
|
| |
def __post_init__(self):
|
| |
if type(self.updated_at) == str:
|
| |
@@ -29,13 +31,10 @@
|
| |
|
| |
|
| |
class Dataset(abc.ABC):
|
| |
- def __init__(self) -> None:
|
| |
- self._source = ""
|
| |
- self._meta = DatasetMeta(filepath="", updated_at=datetime.min)
|
| |
- self._data = pd.DataFrame()
|
| |
-
|
| |
- def __str__(self) -> str:
|
| |
- return f"Last updated: {self.last_updated}"
|
| |
+ def __init__(self) -> NoReturn:
|
| |
+ self._source: str = ""
|
| |
+ self._meta: DatasetMeta = DatasetMeta(filepath="", updated_at=datetime.min)
|
| |
+ self._data: pd.DataFrame = pd.DataFrame()
|
| |
|
| |
@property
|
| |
def id(self) -> str:
|
| |
@@ -46,6 +45,14 @@
|
| |
raise NotImplementedError
|
| |
|
| |
@property
|
| |
+ def last_updated(self) -> datetime:
|
| |
+ return self._meta.updated_at
|
| |
+
|
| |
+ @property
|
| |
+ def description(self) -> str:
|
| |
+ return ""
|
| |
+
|
| |
+ @property
|
| |
def data(self) -> pd.DataFrame:
|
| |
return self._data
|
| |
|
| |
@@ -57,10 +64,6 @@
|
| |
def y(self) -> Union[pd.Series, str]:
|
| |
raise NotImplementedError
|
| |
|
| |
- @property
|
| |
- def last_updated(self) -> datetime:
|
| |
- return self._meta.updated_at
|
| |
-
|
| |
@abc.abstractmethod
|
| |
def _prepare_data(self, dataframe: pd.DataFrame) -> pd.DataFrame:
|
| |
# It should be implemented by the inheritance classes
|
| |
@@ -86,7 +89,7 @@
|
| |
|
| |
return updated_status, meta
|
| |
|
| |
- def load(self, source: str) -> None:
|
| |
+ def load(self, source: str) -> NoReturn:
|
| |
self._source = source
|
| |
meta_updated, meta = self._get_meta()
|
| |
|
| |
@@ -96,7 +99,7 @@
|
| |
self._meta = meta
|
| |
self._data = self._get_data()
|
| |
|
| |
- def update(self):
|
| |
+ def update(self) -> NoReturn:
|
| |
# TODO: Add logging to check when there is an update request
|
| |
meta_updated, meta = self._get_meta()
|
| |
if meta_updated:
|
| |
@@ -104,8 +107,10 @@
|
| |
self._data = self._get_data()
|
| |
|
| |
|
| |
+ # TODO: Maybe add a config file (json or yaml) to setup static fields in the dataset presentation
|
| |
+ # (like titles, and descriptions)
|
| |
class ActiveContributors(Dataset):
|
| |
- def __init__(self):
|
| |
+ def __init__(self) -> NoReturn:
|
| |
super().__init__()
|
| |
self.report_frequency = ReportFrequency.WEEKLY.value
|
| |
|
| |
@@ -118,6 +123,14 @@
|
| |
return "Active Contributors"
|
| |
|
| |
@property
|
| |
+ def description(self) -> str:
|
| |
+ return (
|
| |
+ "Stacked graph of contributors with measured activity each week — and at least 13 weeks total in the last "
|
| |
+ "year. “Old school” contributors have been active for longer than two years; new contributors, less than "
|
| |
+ "one. Blue line shows all contributors active this week regardless of amount of other activity."
|
| |
+ )
|
| |
+
|
| |
+ @property
|
| |
def x(self) -> Union[pd.Index, pd.Series, str]:
|
| |
return self._data.index
|
| |
|
| |
@@ -125,13 +138,10 @@
|
| |
def y(self) -> Union[pd.Series, str]:
|
| |
return "ActiveUsers"
|
| |
|
| |
- def load(self, base_url: str) -> None:
|
| |
+ def load(self, base_url: str) -> NoReturn:
|
| |
endpoint = urljoin(base_url, "reports/contributors")
|
| |
super(ActiveContributors, self).load(endpoint)
|
| |
|
| |
def _prepare_data(self, dataframe: pd.DataFrame) -> pd.DataFrame:
|
| |
dataframe.Date = pd.to_datetime(dataframe.Date, format="%m/%d/%y")
|
| |
return dataframe.resample(self.report_frequency, on="Date").sum()
|
| |
-
|
| |
-
|
| |
- active_contributors = ActiveContributors()
|
| |
Type[Dataset]
toDataset
.