controllable timer with example usage

src/_module_template/router.py

@@ -3,9 +3,46 @@ Router endpoints for the _____ module
 
 Endpoints:
 """
-from fastapi import APIRouter
+import threading
 
+from fastapi import APIRouter, Request, HTTPException
+
+from src.utils import create_timer
 
 router = APIRouter(
 	tags=[""],
-)
+)
+
+@router.put("/timer/start")
+async def start_timer(request: Request, interval: int):
+	def example_timer_target():
+		print("ping")
+
+	stop_event = threading.Event()
+	timer = create_timer(func=example_timer_target, interval=interval, stop_event=stop_event)
+
+	timer_tracker = {
+		"ident": "example_timer",
+		"interval": interval,
+		"stop_event": stop_event,
+		"timer": timer
+	}
+	timer.start()
+
+	request.app.state.timers.append(timer_tracker)
+
+
+@router.put("/timer/stop")
+async def stop_timer(request: Request, ident: str):
+	timers = request.app.state.timers
+	print(timers)
+	idx, timer_tracker = next(((i, timer) for i, timer in enumerate(timers) if timer["ident"] == ident), (None, None))
+
+	if not timer_tracker:
+		raise HTTPException(status_code=404, detail="Timer not found")
+
+	timer_tracker["stop_event"].set()
+	timer_tracker["timer"].join()
+	timer_tracker["timer"] = None
+	timers.pop(idx)
+	print(timers)

src/api.py

@@ -4,12 +4,13 @@ This module hooks the routers for the main endpoints into a single router for im
 from fastapi import APIRouter
 
 from src.auth.router import router as auth_router
+from src._module_template.router import router as template_router
 
 
 api_router = APIRouter()
 
 api_router.include_router(auth_router)
-
+api_router.include_router(template_router)
 
 @api_router.get("/healthcheck", include_in_schema=False)
 def healthcheck():

src/main.py

@@ -23,6 +23,11 @@ async def lifespan(_application: FastAPI) -> AsyncGenerator:
 	# Startup
 	yield
 	# Shutdown
+	if hasattr(_application, "timers"):
+		for timer in _application.timers:
+			timer["stop_event"].set()
+			timer["timer"].join()
+			timer["timer"] = None
 
 
 if settings.ENVIRONMENT.is_deployed:
@@ -58,4 +63,6 @@ app.add_middleware(
 
 app.include_router(api_router)
 
+app.state["timers"] = []
+
 print(f"Running in environment: {settings.ENVIRONMENT}")

src/utils.py

@@ -0,0 +1,15 @@
+"""
+Global non-business-logic reusable functions and classes
+
+Exports:
+"""
+import threading
+from typing import Callable
+
+
+def create_timer(func: Callable, interval: int, stop_event: threading.Event):
+	def target():
+		while not stop_event.wait(interval):
+			func()
+
+	return threading.Thread(target=target, daemon=True)