🎨 Add exceptions and annotations

Author: Dragon-GCS

src/exceptions.py

@@ -0,0 +1,22 @@
+class InvalidTrigger(Exception):
+    def __init__(self, trigger: str) -> None:
+        msg = f"Invalid trigger: {trigger}(type: {type(trigger)}"
+        super().__init__(msg)
+
+
+class InvalidJobStore(Exception):
+    def __init__(self, store: str) -> None:
+        msg = f"Invalid job store: {store}"
+        super().__init__(msg)
+
+
+class InvalidExecutor(Exception):
+    def __init__(self, executor: str) -> None:
+        msg = f"Executor type {executor} not supported"
+        super().__init__(msg)
+
+
+class InvalidAction(Exception):
+    def __init__(self, action: str) -> None:
+        msg = f"Invalid action: {action}"
+        super().__init__(msg)

src/routes/executor.py

@@ -10,13 +10,13 @@
 
 from ..scheduler import scheduler
 from ..schema import ExecutorInfo
-from ..shared import frame_page
+from ..shared import Components, frame_page
 
 router = APIRouter(prefix="/job/executor", tags=["executor"])
 
 
 @router.get("", response_model=FastUI, response_model_exclude_none=True)
-def store():
+def store() -> Components:
     executors = [
         ExecutorInfo.model_validate({"alias": alias, "executor": executor})
         for alias, executor in scheduler._executors.items()
@@ -64,7 +64,9 @@ def store():
 
 
 @router.post("/new", response_model=FastUI, response_model_exclude_none=True)
-async def new_executor(new_executor: Annotated[ExecutorInfo, fastui_form(ExecutorInfo)]):
+async def new_executor(
+    new_executor: Annotated[ExecutorInfo, fastui_form(ExecutorInfo)],
+) -> c.Paragraph:
     alias = new_executor.alias
     if new_executor.alias in scheduler._jobstores:
         return c.Paragraph(text=f"Executor({alias=}) already exists.")
@@ -74,10 +76,10 @@ async def new_executor(new_executor: Annotated[ExecutorInfo, fastui_form(Executo
 
 
 @router.post("/remove", response_model=FastUI, response_model_exclude_none=True)
-async def remove_executor(alias: str = Form()):
+async def remove_executor(alias: Annotated[str, Form()]) -> c.Paragraph:
     if alias not in scheduler._executors:
         return c.Paragraph(text=f"Executor({alias=}) not exists.")
-    elif alias == "default":
+    if alias == "default":
         return c.Paragraph(text="Cannot remove default executor.")
     scheduler.remove_executor(alias)
     return c.Paragraph(text=f"Executor({alias=}) removed successfully.")

src/routes/job.py

@@ -4,21 +4,22 @@
 from uuid import uuid4
 
 from fastapi import APIRouter
-from fastui import FastUI
+from fastui import AnyComponent, FastUI
 from fastui import components as c
 from fastui.components.display import DisplayLookup
 from fastui.events import BackEvent, GoToEvent, PageEvent
 from fastui.forms import fastui_form
 
+from ..exceptions import InvalidAction
 from ..scheduler import scheduler
 from ..schema import JobInfo, ModifyJobParam, NewJobParam
-from ..shared import confirm_modal, frame_page
+from ..shared import Components, confirm_modal, frame_page
 
 router = APIRouter(prefix="/job", tags=["job"])
 
 
 @router.get("/", response_model=FastUI, response_model_exclude_none=True)
-async def jobs():
+async def jobs() -> Components:
     jobs = scheduler.get_jobs()
 
     return frame_page(
@@ -54,7 +55,7 @@ async def jobs():
 
 
 @router.post("/", response_model=FastUI, response_model_exclude_none=True)
-async def new_job(job_info: Annotated[NewJobParam, fastui_form(NewJobParam)]):
+async def new_job(job_info: Annotated[NewJobParam, fastui_form(NewJobParam)]) -> Components:
     trigger = job_info.get_trigger()
     job = scheduler.add_job(
         job_info.func,
@@ -77,7 +78,7 @@ async def new_job(job_info: Annotated[NewJobParam, fastui_form(NewJobParam)]):
 
 
 @router.get("/detail/{id}", response_model=FastUI, response_model_exclude_none=True)
-async def job_detail(id: str):
+async def job_detail(id: str) -> Components:
     job = scheduler.get_job(id)
     if not job:
         return [c.FireEvent(event=GoToEvent(url="/"))]
@@ -134,7 +135,9 @@ async def job_detail(id: str):
 
 
 @router.post("/modify/{id}", response_model=FastUI, response_model_exclude_none=True)
-async def modify_job(id: str, job_info: Annotated[ModifyJobParam, fastui_form(ModifyJobParam)]):
+async def modify_job(
+    id: str, job_info: Annotated[ModifyJobParam, fastui_form(ModifyJobParam)]
+) -> Components:
     modify_kwargs = job_info.model_dump(exclude={"trigger", "trigger_params"})
     modify_kwargs["trigger"] = job_info.get_trigger()
     modify_kwargs = dict(filter(lambda x: x[1], modify_kwargs.items()))
@@ -151,10 +154,12 @@ async def modify_job(id: str, job_info: Annotated[ModifyJobParam, fastui_form(Mo
 
 
 @router.post("/{action}/{id}", response_model=FastUI, response_model_exclude_none=True)
-async def pause_job(action: Literal["pause", "resume", "modify", "reload", "remove"], id: str):
+async def pause_job(
+    action: Literal["pause", "resume", "modify", "reload", "remove"], id: str
+) -> Components:
     job = scheduler.get_job(id)
     if not job:
-        return c.Error(title="Error", description=f"Job({id=}) not found", status_code=500)
+        return [c.Error(title="Error", description=f"Job({id=}) not found", status_code=500)]
 
     match action:
         case "pause":
@@ -167,7 +172,7 @@ async def pause_job(action: Literal["pause", "resume", "modify", "reload", "remo
             module = import_module(job.func.__module__)
             reload(module)
         case _:
-            raise ValueError(f"Invalid action {action}")
+            raise InvalidAction(action)
 
     return [
         c.Paragraph(text=f"Job({id=}, name='{job.name}'), {action=} success."),
@@ -176,7 +181,7 @@ async def pause_job(action: Literal["pause", "resume", "modify", "reload", "remo
 
 
 @router.get("/view/{path:path}", response_model=FastUI, response_model_exclude_none=True)
-def edit_job_script(path: Path):
+def edit_job_script(path: Path) -> AnyComponent:
     if not path.exists():
         return c.Error(title="Error", description=f"Module {path} not found", status_code=500)
     return c.Code(text=path.read_text(), language="python")

src/routes/job_log.py

@@ -1,4 +1,3 @@
-from functools import lru_cache
 from typing import Annotated, Literal
 
 from fastapi import APIRouter
@@ -11,22 +10,22 @@
 
 from ..config import LOG_PATH
 from ..log import PARSE_PATTERN, logger
-from ..shared import frame_page
+from ..shared import Components, frame_page
 from .api import get_available_job_logs
 
 router = APIRouter(prefix="/job/log", tags=["job_log"])
 
 PAGE_LINE = 1000
 
 
-def parse_log_message(text: str):
+def parse_log_message(text: str) -> str:
     messages = text.strip().split("\n", 1)
     if len(messages) > 1:
         return f"{messages[0]}\n```\n{messages[1]}\n```"
     return messages[0]
 
 
-def get_log_content(log_file: str, level: str):
+def get_log_content(log_file: str, level: str) -> list[str]:
     return [
         f"**[{line['pid']}] {line['time']}** *{line['level']}* "
         f"`{line['name']}:{line['line']}`: {parse_log_message(line['message'])}"
@@ -41,15 +40,15 @@ async def get_log(
     log_file: str | None = None,
     level: str = "",
     page: Annotated[int, Field(ge=1)] = 1,
-):
+) -> Components:
     form_fields: list[FormField] = [
         FormFieldSelect(
             title="Level",
             name="level",
             placeholder="Filter by level",
             options=[
                 {"value": level, "label": level}
-                for level in logger._core.levels.keys()  # type: ignore
+                for level in logger._core.levels  # type: ignore
             ],
         )
     ]
@@ -74,7 +73,7 @@ async def get_log(
         log_file = "scheduler.log"
 
     if not (log_file and (LOG_PATH / log_file).exists()):
-        return c.Error(title="File not found", description=f"Log file {log_file} not found.")
+        return [c.Error(title="File not found", description=f"Log file {log_file} not found.")]
 
     contents = get_log_content(log_file, level)
     return frame_page(

src/routes/job_store.py

@@ -10,13 +10,13 @@
 
 from ..scheduler import scheduler
 from ..schema import JobStoreInfo
-from ..shared import frame_page
+from ..shared import Components, frame_page
 
 router = APIRouter(prefix="/job/store", tags=["job_store"])
 
 
 @router.get("", response_model=FastUI, response_model_exclude_none=True)
-def store():
+def store() -> Components:
     job_stores = [
         JobStoreInfo.model_validate({"alias": alias, "store": store})
         for alias, store in scheduler._jobstores.items()
@@ -64,7 +64,9 @@ def store():
 
 
 @router.post("/new", response_model=FastUI, response_model_exclude_none=True)
-async def new_job_store(new_store: Annotated[JobStoreInfo, fastui_form(JobStoreInfo)]):
+async def new_job_store(
+    new_store: Annotated[JobStoreInfo, fastui_form(JobStoreInfo)],
+) -> c.Paragraph:
     alias = new_store.alias
     if new_store.alias in scheduler._jobstores:
         return c.Paragraph(text=f"Job store({alias=}) already exists")
@@ -74,10 +76,10 @@ async def new_job_store(new_store: Annotated[JobStoreInfo, fastui_form(JobStoreI
 
 
 @router.post("/remove", response_model=FastUI, response_model_exclude_none=True)
-async def remove_job_store(alias: str = Form()):
+async def remove_job_store(alias: Annotated[str, Form()]) -> c.Paragraph:
     if alias not in scheduler._jobstores:
         return c.Paragraph(text=f"Job store({alias=}) not exists")
-    elif alias == "default":
+    if alias == "default":
         return c.Paragraph(text="Cannot remove default job store")
     scheduler.remove_jobstore(alias)
     return c.Paragraph(text=f"Job store({alias=}) removed successfully")

src/scheduler.py

@@ -1,5 +1,5 @@
 from functools import partial
-from typing import Literal
+from typing import TYPE_CHECKING, Literal
 
 from apscheduler.events import (
     EVENT_EXECUTOR_ADDED,
@@ -18,43 +18,45 @@
     JobSubmissionEvent,
     SchedulerEvent,
 )
-from apscheduler.job import Job
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
 from .config import SCHEDULER_CONFIG
 from .log import server_log
 
+if TYPE_CHECKING:
+    from apscheduler.job import Job
+
 scheduler = AsyncIOScheduler(**SCHEDULER_CONFIG)
 
 
 def listen_executor_or_jobstore_event(
     event: SchedulerEvent,
     mapper: dict,
     action: Literal["Add executor", "Remove executor", "Add job store", "Remove job store"],
-):
+) -> None:
     obj = f"{event.alias}[{mapper[event.alias]}]" if event.alias in mapper else event.alias
     server_log.debug(f"{action} {obj}")
 
 
 def listen_job_event(
     event: JobEvent, action: Literal["Add job", "Remove job", "Modify job", "Submit job"]
-):
+) -> None:
     job: Job | None = scheduler.get_job(event.job_id)
     server_log.debug(f"{action}: {job.name if job else ''}[{event.job_id}]")
 
 
 def listen_job_execution_event(
     event: JobExecutionEvent,
     action: Literal["Executed job", "Missed job", "Error job"],
-):
+) -> None:
     message = f"{action}: {event.job_id}[{scheduler.get_job(event.job_id)}]"
     if event.exception:
         server_log.opt(exception=event.exception).error(message)
     else:
         server_log.debug(message)
 
 
-def listen_job_submission_event(event: JobSubmissionEvent):
+def listen_job_submission_event(event: JobSubmissionEvent) -> None:
     job: Job | None = scheduler.get_job(event.job_id)
     if not job:
         return