diff --git a/app/cli/automate.py b/app/cli/automate.py index 3691372..48581b6 100644 --- a/app/cli/automate.py +++ b/app/cli/automate.py @@ -3,6 +3,7 @@ import datetime import json import logging from traceback import TracebackException +from typing import Type, TYPE_CHECKING, Any from app import app from app.extensions import db @@ -28,6 +29,11 @@ from app.terraform.proxy.azure_cdn import ProxyAzureCdnAutomation from app.terraform.proxy.cloudfront import ProxyCloudfrontAutomation +if TYPE_CHECKING: + _SubparserType = argparse._SubParsersAction[argparse.ArgumentParser] +else: + _SubparserType = Any + jobs = { x.short_name: x for x in [ @@ -52,17 +58,18 @@ jobs = { } -def run_all(**kwargs): +def run_all(**kwargs: bool) -> None: for job in jobs.values(): run_job(job, **kwargs) -def run_job(job: BaseAutomation, *, force: bool = False, ignore_schedule: bool = False): - automation = Automation.query.filter(Automation.short_name == job.short_name).first() +def run_job(job_cls: Type[BaseAutomation], *, + force: bool = False, ignore_schedule: bool = False) -> None: + automation = Automation.query.filter(Automation.short_name == job_cls.short_name).first() if automation is None: automation = Automation() - automation.short_name = job.short_name - automation.description = job.description + automation.short_name = job_cls.short_name + automation.description = job_cls.description automation.enabled = False automation.next_is_full = False automation.added = datetime.datetime.utcnow() @@ -78,11 +85,11 @@ def run_job(job: BaseAutomation, *, force: bool = False, ignore_schedule: bool = logging.warning("Not time to run this job yet") return if not automation.enabled and not force: - logging.warning(f"job {job.short_name} is disabled and --force not specified") + logging.warning(f"job {job_cls.short_name} is disabled and --force not specified") return automation.state = AutomationState.RUNNING db.session.commit() - job = job() + job: BaseAutomation = job_cls() try: success, logs = job.automate() except Exception as e: @@ -114,7 +121,7 @@ def run_job(job: BaseAutomation, *, force: bool = False, ignore_schedule: bool = class AutomateCliHandler: @classmethod - def add_subparser_to(cls, subparsers: argparse._SubParsersAction) -> None: + def add_subparser_to(cls, subparsers: _SubparserType) -> None: parser = subparsers.add_parser("automate", help="automation operations") parser.add_argument("-a", "--all", dest="all", help="run all automation jobs", action="store_true") parser.add_argument("-j", "--job", dest="job", choices=sorted(jobs.keys()), @@ -123,10 +130,10 @@ class AutomateCliHandler: parser.add_argument("--ignore-schedule", help="run job even if it's not time yet", action="store_true") parser.set_defaults(cls=cls) - def __init__(self, args): + def __init__(self, args: argparse.Namespace) -> None: self.args = args - def run(self): + def run(self) -> None: with app.app_context(): if self.args.job: run_job(jobs[self.args.job], force=self.args.force, ignore_schedule=self.args.ignore_schedule) diff --git a/app/models/__init__.py b/app/models/__init__.py index a448384..a84cfc1 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -1,4 +1,5 @@ from datetime import datetime +from typing import Union, List, Optional, Any from app.extensions import db @@ -38,31 +39,44 @@ class AbstractResource(db.Model): deprecation_reason = db.Column(db.String(), nullable=True) destroyed = db.Column(db.DateTime(), nullable=True) - def __init__(self, **kwargs): - super().__init__(**kwargs) + def __init__(self, *, + id: Optional[int] = None, + added: Optional[datetime] = None, + updated: Optional[datetime] = None, + deprecated: Optional[datetime] = None, + deprecation_reason: Optional[str] = None, + destroyed: Optional[datetime] = None, + **kwargs: Any) -> None: + super().__init__(id=id, + added=added, + updated=updated, + deprecated=deprecated, + deprecation_reason=deprecation_reason, + destroyed=destroyed, + **kwargs) if self.added is None: self.added = datetime.utcnow() if self.updated is None: self.updated = datetime.utcnow() - def deprecate(self, *, reason: str): + def deprecate(self, *, reason: str) -> None: self.deprecated = datetime.utcnow() self.deprecation_reason = reason self.updated = datetime.utcnow() - def destroy(self): + def destroy(self) -> None: if self.deprecated is None: self.deprecated = datetime.utcnow() self.destroyed = datetime.utcnow() self.updated = datetime.utcnow() @classmethod - def csv_header(cls): + def csv_header(cls) -> List[str]: return [ "id", "added", "updated", "deprecated", "deprecation_reason", "destroyed" ] - def csv_row(self): + def csv_row(self) -> List[Union[datetime, bool, int, str]]: return [ getattr(self, x) for x in self.csv_header() ] diff --git a/app/models/activity.py b/app/models/activity.py index 9a4b87b..f5f7523 100644 --- a/app/models/activity.py +++ b/app/models/activity.py @@ -1,4 +1,5 @@ import datetime +from typing import Any, Optional import requests @@ -13,14 +14,25 @@ class Activity(db.Model): text = db.Column(db.Text(), nullable=False) added = db.Column(db.DateTime(), nullable=False) - def __init__(self, **kwargs): - if type(kwargs["activity_type"]) != str or len(kwargs["activity_type"]) > 20 or kwargs["activity_type"] == "": + def __init__(self, *, + id: Optional[int] = None, + group_id: Optional[int] = None, + activity_type: str, + text: str, + added: Optional[datetime.datetime] = None, + **kwargs: Any) -> None: + if type(activity_type) != str or len(activity_type) > 20 or activity_type == "": raise TypeError("expected string for activity type between 1 and 20 characters") - if type(kwargs["text"]) != str: + if type(text) != str: raise TypeError("expected string for text") - if "added" not in kwargs: - kwargs["added"] = datetime.datetime.utcnow() - super().__init__(**kwargs) + super().__init__(id=id, + group_id=group_id, + activity_type=activity_type, + text=text, + added=added, + **kwargs) + if self.added is None: + self.added = datetime.datetime.utcnow() def notify(self) -> int: count = 0