Chore: Formatting and linting

.github/workflows/main.yml

@@ -78,13 +78,12 @@ jobs:
         pip install "setuptools>=64" --upgrade
 
         # Install package in editable mode.
-        # pip install --use-pep517 --prefer-binary --editable=.[test,develop]
-        pip install -r requirements.txt -r requirements-dev.txt
+        pip install --prefer-binary -r requirements.txt -r requirements-dev.txt
+        pip install --prefer-binary --editable=.[test,develop]
 
     - name: Run linter and software tests
       run: |
-        # poe check
-        poe test
+        poe check
 
     - name: Upload coverage to Codecov
       uses: codecov/codecov-action@v3

README.md

@@ -38,7 +38,8 @@ pinning is strongly recommended, especially when you use it as a library.
 ## Setup
 
 ```shell
-pip install -r requirements.txt -r requirements-dev.txt
+pip install --upgrade --prefer-binary -r requirements.txt -r requirements-dev.txt
+pip install --upgrade --prefer-binary --editable=.
 ```
 
 
@@ -101,7 +102,7 @@ documentation].
 
 [APScheduler]: https://pypi.org/project/APScheduler/
 [cron]: https://en.wikipedia.org/wiki/Cron
-[development sandbox documentation]: https://github.com/WalBeh/scheduler-playground/blob/main/docs/sandbox.md
+[development sandbox documentation]: https://github.com/WalBeh/scheduler-playground/blob/master/docs/sandbox.md
 [DWIM]: https://en.wikipedia.org/wiki/DWIM
 [FastAPI]: https://pypi.org/project/fastapi/
 [Pydantic]: https://pypi.org/project/pydantic/

cli.py

@@ -1,12 +1,12 @@
 import click
-from dotenv import load_dotenv, find_dotenv
+from dotenv import find_dotenv, load_dotenv
 
 from core import Supertask
 from job_seeder import JobSeeder
 from models import JobStoreLocation
 from util import setup_logging
 
-
+# TODO: Gate behind an environment variable?
 load_dotenv(find_dotenv())
 
 

core.py

@@ -1,35 +1,40 @@
 import logging
+import os
+import threading
+import time
+import typing as t
+
 import icecream
+import pytz
+import uvicorn
+from apscheduler.executors.pool import ProcessPoolExecutor, ThreadPoolExecutor
+from apscheduler.jobstores.memory import MemoryJobStore
 from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
 from apscheduler.schedulers.background import BackgroundScheduler
-from apscheduler.jobstores.memory import MemoryJobStore
-from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
-from icecream import ic
-import typing as t
-import time
-import uvicorn
-import os
-import pytz
-import threading
+from fastapi import FastAPI
 from halo import Halo
+from icecream import ic
 
+from cronjob_routes import router as cronjob_router
+from jobstore_sqlalchemy import CrateDBSQLAlchemyJobStore
 from models import JobStoreLocation
 from settings import Settings
-from jobstore_sqlalchemy import CrateDBSQLAlchemyJobStore
-from fastapi import FastAPI
-from cronjob_routes import router as cronjob_router
 
 logger = logging.getLogger(__name__)
 
 icecream.IceCreamDebugger.lineWrapWidth = 120
 
 
 class Supertask:
-
     SQLALCHEMY_ECHO = False
 
-    def __init__(self, store: t.Union[JobStoreLocation, str], pre_delete_jobs: bool = False, pre_seed_jobs: str = None, debug: bool = False):
-
+    def __init__(
+        self,
+        store: t.Union[JobStoreLocation, str],
+        pre_delete_jobs: bool = False,
+        pre_seed_jobs: str = None,
+        debug: bool = False,
+    ):
         # Bundle settings to be able to propagate them to the FastAPI subsystem.
         if isinstance(store, str):
             store = JobStoreLocation(address=store)
@@ -58,36 +63,35 @@ def configure(self):
             # TODO: Need to run `CREATE SCHEMA ...` before using it?
             job_store = SQLAlchemyJobStore(url=address, tablename=table, engine_options={"echo": self.SQLALCHEMY_ECHO})
         elif address.startswith("crate://"):
-            job_store = CrateDBSQLAlchemyJobStore(url=address, tableschema=schema, tablename=table, engine_options={"echo": self.SQLALCHEMY_ECHO})
+            job_store = CrateDBSQLAlchemyJobStore(
+                url=address, tableschema=schema, tablename=table, engine_options={"echo": self.SQLALCHEMY_ECHO}
+            )
         else:
             raise RuntimeError(f"Initializing job store failed. Unknown address: {address}")
 
         if self.settings.pre_delete_jobs:
             try:
                 job_store.remove_all_jobs()
-            except:
+            except Exception:  # noqa: S110
                 pass
 
-        job_defaults = {
-            'coalesce': False,
-            'max_instances': 1
-        }
-        executors = {
-            'default': ThreadPoolExecutor(20),
-            'processpool': ProcessPoolExecutor(5)
-        }
+        job_defaults = {"coalesce": False, "max_instances": 1}
+        executors = {"default": ThreadPoolExecutor(20), "processpool": ProcessPoolExecutor(5)}
         job_stores = {
-            'default': job_store,
+            "default": job_store,
         }
 
         # Create a timezone object for Vienna
-        timezone = pytz.timezone('Europe/Vienna')
-        self.scheduler = BackgroundScheduler(executors=executors, job_defaults=job_defaults, jobstores=job_stores, timezone=timezone)
-        logger.info(f"Configured scheduler: "
-                    f"executors={self.scheduler._executors}, "
-                    f"jobstores={self.scheduler._jobstores}, "
-                    f"timezone={self.scheduler.timezone}"
-                    )
+        timezone = pytz.timezone("Europe/Vienna")
+        self.scheduler = BackgroundScheduler(
+            executors=executors, job_defaults=job_defaults, jobstores=job_stores, timezone=timezone
+        )
+        logger.info(
+            f"Configured scheduler: "
+            f"executors={self.scheduler._executors}, "
+            f"jobstores={self.scheduler._jobstores}, "
+            f"timezone={self.scheduler.timezone}"
+        )
         return self
 
     def start(self, listen_http: str = None):
@@ -100,7 +104,7 @@ def start_scheduler(self):
         logger.info("Starting scheduler")
         self.scheduler.start()
         start = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
-        ic('//======= START ======', start)
+        ic("//======= START ======", start)
 
         # Get next run time for all jobs
         jobs = self.scheduler.get_jobs()
@@ -109,8 +113,8 @@ def start_scheduler(self):
         return self
 
     def wait(self):
-        print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))
-        spinner = Halo(text='Waiting', spinner='dots')
+        print("Press Ctrl+{0} to exit".format("Break" if os.name == "nt" else "C"))  # noqa: T201
+        spinner = Halo(text="Waiting", spinner="dots")
         spinner.start()
         try:
             # This is here to simulate application activity (which keeps the main thread alive).
@@ -122,7 +126,6 @@ def wait(self):
         return self
 
     def start_http_service(self, listen_http: str):
-
         host, port_str = listen_http.split(":")
         port = int(port_str)
 
@@ -134,7 +137,6 @@ def start_http_service(self, listen_http: str):
         app.dependency_overrides[Settings] = lambda: self.settings
 
         app.include_router(cronjob_router)
-        #app.include_router(cronjob_router, dependencies=[Depends(JsonResourceWithAddress(self.pre_seed_jobs))])
 
         def run_server():
             uvicorn.run(app, host=host, port=port)

cronjob_routes.py

@@ -1,13 +1,17 @@
+# ruff: noqa: B008
+import logging
 import typing as t
 
-from fastapi import APIRouter, HTTPException, Depends, Form
-from fastapi.templating import Jinja2Templates
+import fastapi.responses
+from fastapi import APIRouter, Depends, Form, HTTPException, Request
 from fastapi.responses import HTMLResponse
-from fastapi import Request
+from fastapi.templating import Jinja2Templates
 
-from settings import Settings
-from models import CronJob
 from database import JsonResource
+from models import CronJob
+from settings import Settings
+
+logger = logging.getLogger(__name__)
 
 router = APIRouter()
 templates = Jinja2Templates(directory="templates")
@@ -18,6 +22,11 @@ def get_json_resource(settings: Settings = Depends()) -> JsonResource:
     FastAPI Dependency to provide a JsonResource instance to the request handlers.
     """
     from database import JsonResource
+
+    if settings.pre_seed_jobs is None:
+        msg = "No web UI without pre-seed file"
+        logger.error(msg)
+        raise fastapi.exceptions.HTTPException(status_code=400, detail=msg)
     return JsonResource(filepath=settings.pre_seed_jobs)
 
 
@@ -26,26 +35,35 @@ async def jobs_page(request: Request, json_resource: JsonResource = Depends(get_
     jobs = json_resource.read_index()
     return templates.TemplateResponse("jobs.html", {"request": request, "jobs": jobs})
 
+
 @router.post("/cronjobs/", response_model=CronJob)
-def create_cronjob(crontab: str = Form(...), job: str = Form(...), enabled: bool = Form(False), json_resource: JsonResource = Depends(get_json_resource)):
+def create_cronjob(
+    crontab: str = Form(...),
+    job: str = Form(...),
+    enabled: bool = Form(False),
+    json_resource: JsonResource = Depends(get_json_resource),
+):
     db = json_resource.read()
-    cronjob = CronJob(id=len(db)+1, crontab=crontab, job=job, enabled=enabled, last_run=None, last_status=None)
+    cronjob = CronJob(id=len(db) + 1, crontab=crontab, job=job, enabled=enabled, last_run=None, last_status=None)
     db.append(cronjob)
     json_resource.write(db)
     return cronjob
 
+
 @router.get("/cronjobs/", response_model=t.List[CronJob])
 def read_cronjobs(json_resource: JsonResource = Depends(get_json_resource)):
     db = json_resource.read()
     return db
 
+
 @router.get("/cronjobs/{cronjob_id}", response_model=CronJob)
 def read_cronjob(cronjob_id: int, json_resource: JsonResource = Depends(get_json_resource)):
     db = json_resource.read()
     if cronjob_id < 0 or cronjob_id >= len(db):
         raise HTTPException(status_code=404, detail="CronJob not found")
     return db[cronjob_id]
 
+
 @router.put("/cronjobs/{cronjob_id}", response_model=CronJob)
 def update_cronjob(cronjob_id: int, cronjob: CronJob, json_resource: JsonResource = Depends(get_json_resource)):
     db = json_resource.read()
@@ -55,6 +73,7 @@ def update_cronjob(cronjob_id: int, cronjob: CronJob, json_resource: JsonResourc
     json_resource.write(db)
     return cronjob
 
+
 @router.delete("/cronjobs/{cronjob_id}", response_model=CronJob)
 def delete_cronjob(cronjob_id: int, json_resource: JsonResource = Depends(get_json_resource)):
     db = json_resource.read()

database.py

@@ -1,41 +1,45 @@
 import json
+import typing as t
 
 from pueblo.io import to_io
 
 from models import CronJob
 
 # Sample data storex
-cronjobs_db = []
-# # Sample data store
-# sample_cronjobs_data = [
-#     {"cron": "*/5 * * * *", "job": "job1", "enabled": True},
-#     {"cron": "0 * * * *", "job": "job2", "enabled": True},
-#     {"cron": "*/15 * * * *", "job": "job3", "enabled": False},
-# ]
+cronjobs_db: t.List[CronJob] = []
 
-# cronjobs_db = [CronJob(**cronjob) for cronjob in sample_cronjobs_data]
 
+"""
+# Sample data store
+sample_cronjobs_data = [
+    {"cron": "*/5 * * * *", "job": "job1", "enabled": True},
+    {"cron": "0 * * * *", "job": "job2", "enabled": True},
+    {"cron": "*/15 * * * *", "job": "job3", "enabled": False},
+]
+
+cronjobs_db = [CronJob(**cronjob) for cronjob in sample_cronjobs_data]
+"""
 
-class JsonResource:
 
+class JsonResource:
     def __init__(self, filepath: str):
         self.filepath = filepath
 
     def read_index(self):
-        with open(self.filepath, 'r') as f:
+        with open(self.filepath, "r") as f:
             cronjobs_data = json.load(f)
         return [CronJob(**job) for job in cronjobs_data]
 
     def read(self):
-        with to_io(self.filepath, 'r') as f:
+        with to_io(self.filepath, "r") as f:
             cronjobs_data = json.load(f)
         for cronjob in cronjobs_data:
-            if 'id' in cronjob:
-                del cronjob['id']
+            if "id" in cronjob:
+                del cronjob["id"]
         cronjobs_db = [CronJob(id=i, **cronjob) for i, cronjob in enumerate(cronjobs_data)]
         return cronjobs_db
 
     def write(self, db):
         cronjobs_data = [cronjob.dict() for cronjob in db]
-        with open(self.filepath, 'w') as f:
+        with open(self.filepath, "w") as f:
             json.dump(cronjobs_data, f)