feat: add AppMetaclass & background sleep

Author: NightMarcher

examples/service/__init__.py

@@ -33,6 +33,11 @@
  "level": "DEBUG",
  "propagate": False,
  },
+ "examples.service.routers.basic": {
+ "handlers": ["console"],
+ "level": "DEBUG",
+ "propagate": False,
+ },
  }
 })
 

examples/service/models/demo/demo_meta.py

@@ -4,10 +4,8 @@
 
 class DemoMetaclass(AppMetaclass):
 
- @property
- def ro_conn(self):
+ def get_ro_conn(self):
  return Tortoise.get_connection("demo_ro")
 
- @property
- def rw_conn(self):
+ def get_rw_conn(self):
  return Tortoise.get_connection("demo_rw")

examples/service/routers/basic.py

@@ -1,8 +1,34 @@
-from fastapi import APIRouter
+from asyncio import sleep
+import logging
+from time import perf_counter
+
+from fastapi import APIRouter, BackgroundTasks
 
 router = APIRouter()
+logger = logging.getLogger(__name__)
 
 
 @router.get("/ping")
-async def pong():
+async def ping():
  return {"ping": "pong"}
+
+
+@router.get("/background/sleep")
+async def bg_sleep(
+ bg_tasks: BackgroundTasks,
+):
+ t0 = perf_counter()
+ # This sleep represents time-consuming procedure
+ await sleep(1)
+
+ t1 = perf_counter()
+ async def my_sleep(sec):
+ await sleep(sec)
+ logger.debug(f"slept {sec} sec")
+ bg_tasks.add_task(my_sleep, 1)
+ bg_tasks.add_task(my_sleep, 1.5)
+ bg_tasks.add_task(my_sleep, 2)
+
+ t2 = perf_counter()
+ logger.debug(f"{1000 * (t1 - t0):.3f} ms, {1000 * (t2 - t0):.3f} ms")
+ return "done"

fastapi_esql/orm/base_app.py

@@ -1,18 +1,22 @@
-from abc import abstractproperty
+from abc import ABCMeta
 
 
-class AppMetaclass(type):
+class AppMetaclass(ABCMeta):
 
- @abstractproperty
+ @property
  def ro_conn(self):
  """
  return Tortoise.get_connection("ro_conn")
  """
- raise NotImplementedError(f"{self.__class__.__name__}`s property method 'ro_conn' was not implemented!")
+ if not getattr(self, "get_ro_conn", None):
+ raise NotImplementedError(f"Method get_ro_conn() was not implemented by {self.__class__.__name__}!")
+ return self.get_ro_conn()
 
- @abstractproperty
+ @property
  def rw_conn(self):
  """
  return Tortoise.get_connection("rw_conn")
  """
- raise NotImplementedError(f"{self.__class__.__name__}`s property method 'rw_conn' was not implemented!")
+ if not getattr(self, "get_rw_conn", None):
+ raise NotImplementedError(f"Method get_rw_conn() was not implemented by {self.__class__.__name__}!")
+ return self.get_rw_conn()