Adds scripts for scheduled tasks providing interface for Schedule API

Author: caseneuve

.gitignore

@@ -96,3 +96,8 @@ ENV/
 .ropeproject
 .venv
 
+# Emacs local variables
+.dir-locals.el
+
+# pytest
+.pytest_cache/

pythonanywhere/schedule_api.py

@@ -0,0 +1,113 @@
+""" Interface speaking with PythonAnywhere API providing methods for scheduled tasks.
+*Don't use* `Schedule` :class: in helper scripts, use `task.Task` instead."""
+
+import getpass
+
+from pythonanywhere.api import call_api, get_api_endpoint
+
+
+class Schedule:
+    """Interface for PythonAnywhere scheduled tasks API.
+
+    Uses `pythonanywhere.api` :method: `get_api_endpoint` to create url,
+    which is stored in a class variable `Schedule.base_url`, then calls
+    `call_api` with appropriate arguments to execute scheduled tasks tasks
+    actions. Covers 'GET' and 'POST' methods for tasks list, as well as
+    'GET', 'PATCH' and 'DELETE' methods for task with id.
+
+    Use :method: `Schedule.get_list` to get all tasks list.
+    Use :method: `Schedule.create` to create new task.
+    Use :method: `Schedule.get_specs` to get existing task specs.
+    Use :method: `Schedule.delete` to delete existing task.
+    Use :method: `Schedule.update` to update existing task."""
+
+    base_url = get_api_endpoint().format(username=getpass.getuser(), flavor="schedule")
+
+    def get_list(self):
+        """Gets list of existing scheduled tasks.
+
+        :returns: list of existing scheduled tasks specs"""
+
+        return call_api(self.base_url, "GET").json()
+
+    def create(self, params):
+        """Creates new scheduled task using `params`.
+
+        Params should be: command, enabled (True or False), interval (daily or
+        hourly), hour (24h format) and minute.
+
+        :param params: dictionary with required scheduled task specs
+        :returns: dictionary with created task specs"""
+
+        result = call_api(self.base_url, "POST", json=params)
+
+        if result.status_code == 201:
+            return result.json()
+
+        if not result.ok:
+            raise Exception(
+                "POST to set new task via API failed, got {result}: "
+                "{result_text}".format(result=result, result_text=result.text)
+            )
+
+    def get_specs(self, task_id):
+        """Get task specs by id.
+
+        :param task_id: existing task id
+        :returns: dictionary of existing task specs"""
+
+        result = call_api(
+            "{base_url}{task_id}/".format(base_url=self.base_url, task_id=task_id), "GET"
+        )
+        if result.status_code == 200:
+            return result.json()
+        else:
+            raise Exception(
+                "Could not get task with id {task_id}. Got result {result}: {content}".format(
+                    task_id=task_id, result=result, content=result.text
+                )
+            )
+
+    def delete(self, task_id):
+        """Deletes scheduled task by id.
+
+        :param task_id: scheduled task to be deleted id number
+        :returns: True when API response is 204"""
+
+        result = call_api(
+            "{base_url}{task_id}/".format(base_url=self.base_url, task_id=task_id), "DELETE"
+        )
+
+        if result.status_code == 204:
+            return True
+
+        if not result.ok:
+            raise Exception(
+                "DELETE via API on task {task_id} failed, got {result}: "
+                "{result_text}".format(task_id=task_id, result=result, result_text=result.text)
+            )
+
+    def update(self, task_id, params):
+        """Updates existing task using id and params.
+
+        Params should at least one of: command, enabled, interval, hour,
+        minute. To update hourly task don't use 'hour' param. On the other
+        hand when changing task's interval from 'hourly' to 'daily' hour is
+        required.
+
+        :param task_id: existing task id
+        :param params: dictionary of specs to update"""
+
+        result = call_api(
+            "{base_url}{task_id}/".format(base_url=self.base_url, task_id=task_id),
+            "PATCH",
+            json=params,
+        )
+        if result.status_code == 200:
+            return result.json()
+        else:
+            raise Exception(
+                "Could not update task {task_id}. Got {result}: {content}".format(
+                    task_id=task_id, result=result, content=result.text
+                )
+            )

pythonanywhere/schedule_api.pyi

@@ -0,0 +1,11 @@
+from typing import List, Optional
+
+from typing_extensions import Literal
+
+class Schedule:
+    base_url: str = ...
+    def get_list(self) -> List[dict]: ...
+    def create(self, params: dict) -> Optional[dict]: ...
+    def get_specs(self, task_id: int) -> dict: ...
+    def delete(self, task_id: int) -> Literal[True]: ...
+    def update(self, task_id: int, params: dict) -> dict: ...

pythonanywhere/scripts_commons.py

@@ -0,0 +1,123 @@
+"""Helpers used by pythonanywhere helper scripts."""
+
+import logging
+import sys
+
+from schema import And, Or, Schema, SchemaError, Use
+
+from pythonanywhere.snakesay import snakesay
+from pythonanywhere.task import Task
+
+logger = logging.getLogger(__name__)
+
+# fmt: off
+tabulate_formats = [
+    "plain", "simple", "github", "grid", "fancy_grid", "pipe", "orgtbl", "jira",
+    "presto", "psql", "rst", "mediawiki", "moinmoin", "youtrack", "html", "latex",
+    "latex_raw", "latex_booktabs", "textile",
+]
+# fmt: on
+
+
+class ScriptSchema(Schema):
+    """Extends `Schema` adapting it to PA scripts validation strategies.
+
+    Adds predefined schemata as class variables to be used in scripts'
+    validation schemas as well as `validate_user_input` method which acts
+    as `Schema.validate` but returns a dictionary with converted keys
+    ready to be used as function keyword arguments, e.g. validated
+    arguments {"--foo": bar, "<baz>": qux} will be be converted to
+    {"foo": bar, "baz": qux}. Additional conversion rules may be added as
+    dictionary passed to `validate_user_input` :method: as `conversions`
+    :param:.
+
+    Use :method:`ScriptSchema.validate_user_input` to obtain kwarg
+    dictionary."""
+
+    # class variables are used in task scripts schemata:
+    boolean = Or(None, bool)
+    hour = Or(None, And(Use(int), lambda h: 0 <= h <= 23), error="--hour has to be in 0..23")
+    id_multi = Or([], And(lambda y: [x.isdigit() for x in y], error="<id> has to be integer"))
+    id_required = And(Use(int), error="<id> has to be an integer")
+    minute_required = And(Use(int), lambda m: 0 <= m <= 59, error="--minute has to be in 0..59")
+    minute = Or(None, minute_required)
+    string = Or(None, str)
+    tabulate_format = Or(
+        None,
+        And(str, lambda f: f in tabulate_formats),
+        error="--format should match one of: {}".format(", ".join(tabulate_formats)),
+    )
+
+    replacements = {"--": "", "<": "", ">": ""}
+
+    def convert(self, string):
+        """Removes cli argument notation characters ('--', '<', '>' etc.).
+
+        :param string: cli argument key to be converted to fit Python
+        argument syntax."""
+
+        for key, value in self.replacements.items():
+            string = string.replace(key, value)
+        return string
+
+    def validate_user_input(self, arguments, *, conversions=None):
+        """Calls `Schema.validate` on provided `arguments`.
+
+        Returns dictionary with keys converted by
+        `ScriptSchema.convert` :method: to be later used as kwarg
+        arguments. Universal rules for conversion are stored in
+        `replacements` class variable and may be updated using
+        `conversions` kwarg. Use optional `conversions` :param: to add
+        custom replacement rules.
+
+        :param arguments: dictionary of cli arguments provided be
+        (e.g.) `docopt`
+        :param conversions: dictionary of additional rules to
+        `self.replacements`"""
+
+        if conversions:
+            self.replacements.update(conversions)
+
+        try:
+            self.validate(arguments)
+            return {self.convert(key): val for key, val in arguments.items()}
+        except SchemaError as e:
+            logger.warning(snakesay(str(e)))
+            sys.exit(1)
+
+
+def get_logger(set_info=False):
+    """Sets logger for 'pythonanywhere' package.
+
+    Returns `logging.Logger` instance with no message formatting which
+    will stream to stdout. With `set_info` :param: set to `True`
+    logger defines `logging.INFO` level otherwise it leaves default
+    `logging.WARNING`.
+
+    To toggle message visibility in scripts use `logger.info` calls
+    and switch `set_info` value accordingly.
+
+    :param set_info: boolean (defaults to False)"""
+
+    logging.basicConfig(format="%(message)s", stream=sys.stdout)
+    logger = logging.getLogger("pythonanywhere")
+    if set_info:
+        logger.setLevel(logging.INFO)
+    else:
+        logger.setLevel(logging.WARNING)
+    return logger
+
+
+def get_task_from_id(task_id, no_exit=False):
+    """Get `Task.from_id` instance representing existing task.
+
+    :param task_id: integer (should be a valid task id)
+    :param no_exit: if (default) False sys.exit will be called when
+      exception is caught"""
+
+    try:
+        return Task.from_id(task_id)
+    except Exception as e:
+        logger.warning(snakesay(str(e)))
+        if not no_exit:
+            sys.exit(1)

pythonanywhere/scripts_commons.pyi

@@ -0,0 +1,25 @@
+import logging
+from typing import Dict, List, Optional
+
+import schema
+
+from pythonanywhere.task import Task
+
+logger: logging.Logger = ...
+tabulate_formats: List[str] = ...
+
+class ScriptSchema(schema.Schema):
+    boolean: schema.Or = ...
+    hour: schema.Or = ...
+    id_multi: schema.Or = ...
+    id_required: schema.And = ...
+    minute: schema.Or = ...
+    minute_required: schema.And = ...
+    string: schema.Or = ...
+    tabulate_format: schema.Or = ...
+    replacements: Dict[str] = ...
+    def convert(self, string: str) -> str: ...
+    def validate_user_input(self, arguments: dict, *, conversions: Optional[dict]) -> dict: ...
+
+def get_logger(set_info: bool) -> logging.Logger: ...
+def get_task_from_id(task_id: int, no_exit: bool) -> Task: ...