working delayed route and scheduled route

Author: ofpiyush

README.md

@@ -30,16 +30,25 @@ If we host the task worker on Cloud Run, we get autoscaling workers.
 In practice, this is what it looks like:
 
 ```python
-router = APIRouter(route_class=TaskRouteBuilder(...))
+delayed_router = APIRouter(route_class=DelayedRouteBuilder(...))
+scheduled_router = APIRouter(route_class=ScheduledRouteBuilder(...))
 
 class Recipe(BaseModel):
     ingredients: List[str]
 
-@router.post("/{restaurant}/make_dinner")
-async def make_dinner(restaurant: str, recipe: Recipe,):
+@delayed_router.post("/{restaurant}/make_dinner")
+async def make_dinner(restaurant: str, recipe: Recipe):
     # Do a ton of work here.
 
-app.include_router(router)
+@scheduled_router.post("/home_cook")
+async def home_cook(recipe: Recipe):
+    # Make my own food
+
+app.include_router(delayed_router)
+app.include_router(scheduled_router)
+
+# If you wan to make your own breakfast every morning at 7AM IST.
+home_cook.scheduler(name="test-home-cook-at-7AM-IST", schedule="0 7 * * *", time_zone="Asia/Kolkata").schedule(recipe=Recipe(ingredients=["Milk","Cereal"]))
 ```
 
 Now we can trigger the task with
@@ -54,11 +63,6 @@ If we want to trigger the task 30 minutes later
 make_dinner.options(countdown=1800).delay(...)
 ```
 
-If we want to schedule it to run every evening at 7PM IST
-
-```python
-make_dinner.scheduler(name="test-make-dinner-at-7PM-IST", schedule="0 19 * * *", time_zone="Asia/Kolkata").schedule(...)
-```
 ## Running
 
 ### Local
@@ -83,9 +87,9 @@ Forwarding                    http://feda-49-207-221-153.ngrok.io -> http://loca
 ```python
 # complete file: examples/simple/main.py
 
-# First we construct our TaskRoute class with all relevant settings
+# First we construct our DelayedRoute class with all relevant settings
 # This can be done once across the entire project
-TaskRoute = TaskRouteBuilder(
+DelayedRoute = DelayedRouteBuilder(
     base_url="http://feda-49-207-221-153.ngrok.io",
     queue_path=queue_path(
         project="gcp-project-id",
@@ -94,13 +98,12 @@ TaskRoute = TaskRouteBuilder(
     ),
 )
 
-# Wherever we use
-task_router = APIRouter(route_class=TaskRoute, prefix="/tasks")
+delayed_router = APIRouter(route_class=DelayedRoute, prefix="/tasks")
 
 class Payload(BaseModel):
     message: str
 
-@task_router.post("/hello")
+@delayed_router.post("/hello")
 async def hello(p: Payload = Payload(message="Default")):
     logger.warning(f"Hello task ran with payload: {p.message}")
 
@@ -114,7 +117,7 @@ async def trigger():
     hello.delay(p=Payload(message="Triggered task"))
     return {"message": "Hello task triggered"}
 
-app.include_router(task_router)
+app.include_router(delayed_router)
 
 ```
 
@@ -156,7 +159,7 @@ We'll only edit the parts from above that we need changed from above example.
 # URL of the Cloud Run service
 base_url = "https://hello-randomchars-el.a.run.app"
 
-TaskRoute = TaskRouteBuilder(
+DelayedRoute = DelayedRouteBuilder(
     base_url=base_url,
     # Task queue, same as above.
     queue_path=queue_path(...),
@@ -174,15 +177,15 @@ Check the fleshed out example at [`examples/full/tasks.py`](examples/full/tasks.
 
 ## Configuration
 
-### TaskRouteBuilder
+### DelayedRouteBuilder
 
 Usage:
 
 ```python
-TaskRoute = TaskRouteBuilder(...)
-task_router = APIRouter(route_class=TaskRoute)
+DelayedRoute = DelayedRouteBuilder(...)
+delayed_router = APIRouter(route_class=DelayedRoute)
 
-@task_router.get("/simple_task")
+@delayed_router.get("/simple_task")
 def mySimpleTask():
     return {}
 ```
@@ -197,7 +200,7 @@ def mySimpleTask():
 
 - `client` - If you need to override the Cloud Tasks client, pass the client here. (eg: changing credentials, transport etc)
 
-### Task level default options
+#### Task level default options
 
 Usage:
 
@@ -225,15 +228,15 @@ def mySimpleTask():
     return {}
 ```
 
-### Delayer Options
+#### Delayer Options
 
 Usage:
 
 ```python
 mySimpleTask.options(...).delay()
 ```
 
-All options from above can be overriden per call (including TaskRouteBuilder options like `base_url`) with kwargs to the `options` function before calling delay.
+All options from above can be overriden per call (including DelayedRouteBuilder options like `base_url`) with kwargs to the `options` function before calling delay.
 
 Example:
 
@@ -242,14 +245,31 @@ Example:
 mySimpleTask.options(countdown=120).delay()
 ```
 
+### ScheduledRouteBuilder
+
+Usage:
+
+```python
+ScheduledRoute = ScheduledRouteBuilder(...)
+scheduled_router = APIRouter(route_class=ScheduledRoute)
+
+@scheduled_router.get("/simple_scheduled_task")
+def mySimpleScheduledTask():
+    return {}
+
+
+mySimpleScheduledTask.scheduler(name="simple_scheduled_task", schedule="* * * * *").schedule()
+```
+
+
 ## Hooks
 
 We might need to override things in the task being sent to Cloud Tasks. The `pre_create_hook` allows us to do that.
 
 Some hooks are included in the library.
 
-- `oidc_task_hook` / `oidc_scheduler_hook` - Used to pass OIDC token (for Cloud Run etc).
-- `deadline_task_hook` / `deadline_scheduler_hook` - Used to change the timeout for the worker of a task. (PS: this deadline is decided by the sender to the queue and not the worker)
+- `oidc_delayed_hook` / `oidc_scheduled_hook` - Used to pass OIDC token (for Cloud Run etc).
+- `deadline_delayed_hook` / `deadline_scheduled_hook` - Used to change the timeout for the worker of a task. (PS: this deadline is decided by the sender to the queue and not the worker)
 - `chained_hook` - If you need to chain multiple hooks together, you can do that with `chained_hook(hook1, hook2)`
 
 ## Future work

examples/full/settings.py

@@ -6,11 +6,13 @@
 from google.cloud import tasks_v2
 
 # Imports from this repository
+from fastapi_cloud_tasks.utils import location_path
 from fastapi_cloud_tasks.utils import queue_path
 
-TASK_LISTENER_BASE_URL = os.getenv("TASK_LISTENER_BASE_URL", default="https://example.com")
+TASK_LISTENER_BASE_URL = os.getenv("TASK_LISTENER_BASE_URL", default="https://65fb-35-207-241-4.ngrok.io")
 TASK_PROJECT_ID = os.getenv("TASK_PROJECT_ID", default="sample-project")
 TASK_LOCATION = os.getenv("TASK_LOCATION", default="asia-south1")
+SCHEDULED_LOCATION = os.getenv("SCHEDULED_LOCATION", default="us-central1")
 TASK_QUEUE = os.getenv("TASK_QUEUE", default="test-queue")
 
 TASK_SERVICE_ACCOUNT = os.getenv(
@@ -24,7 +26,12 @@
     queue=TASK_QUEUE,
 )
 
+SCHEDULED_LOCATION_PATH = location_path(
+    project=TASK_PROJECT_ID,
+    location=SCHEDULED_LOCATION,
+)
+
 TASK_OIDC_TOKEN = tasks_v2.OidcToken(service_account_email=TASK_SERVICE_ACCOUNT, audience=TASK_LISTENER_BASE_URL)
-SCHEDULER_OIDC_TOKEN = scheduler_v1.OidcToken(
+SCHEDULED_OIDC_TOKEN = scheduler_v1.OidcToken(
     service_account_email=TASK_SERVICE_ACCOUNT, audience=TASK_LISTENER_BASE_URL
 )

examples/full/tasks.py

@@ -8,55 +8,65 @@
 
 # Imports from this repository
 from examples.full.serializer import Payload
-from examples.full.settings import SCHEDULER_OIDC_TOKEN
+from examples.full.settings import SCHEDULED_LOCATION_PATH
+from examples.full.settings import SCHEDULED_OIDC_TOKEN
 from examples.full.settings import TASK_LISTENER_BASE_URL
 from examples.full.settings import TASK_OIDC_TOKEN
 from examples.full.settings import TASK_QUEUE_PATH
+from fastapi_cloud_tasks.delayed_route import DelayedRouteBuilder
 from fastapi_cloud_tasks.hooks import chained_hook
-from fastapi_cloud_tasks.hooks import deadline_scheduler_hook
-from fastapi_cloud_tasks.hooks import deadline_task_hook
-from fastapi_cloud_tasks.hooks import oidc_scheduler_hook
-from fastapi_cloud_tasks.hooks import oidc_task_hook
-from fastapi_cloud_tasks.taskroute import TaskRouteBuilder
+from fastapi_cloud_tasks.hooks import deadline_delayed_hook
+from fastapi_cloud_tasks.hooks import deadline_scheduled_hook
+from fastapi_cloud_tasks.hooks import oidc_delayed_hook
+from fastapi_cloud_tasks.hooks import oidc_scheduled_hook
+from fastapi_cloud_tasks.scheduled_route import ScheduledRouteBuilder
 
 app = FastAPI()
 
 
 logger = logging.getLogger("uvicorn")
 
-TaskRoute = TaskRouteBuilder(
+DelayedRoute = DelayedRouteBuilder(
     base_url=TASK_LISTENER_BASE_URL,
     queue_path=TASK_QUEUE_PATH,
     # Chain multiple hooks together
     pre_create_hook=chained_hook(
         # Add service account for cloud run
-        oidc_task_hook(
+        oidc_delayed_hook(
             token=TASK_OIDC_TOKEN,
         ),
         # Wait for half an hour
-        deadline_task_hook(duration=duration_pb2.Duration(seconds=1800)),
+        deadline_delayed_hook(duration=duration_pb2.Duration(seconds=1800)),
     ),
-    pre_scheduler_hook=chained_hook(
+)
+
+ScheduledRoute = ScheduledRouteBuilder(
+    base_url=TASK_LISTENER_BASE_URL,
+    location_path=SCHEDULED_LOCATION_PATH,
+    pre_create_hook=chained_hook(
         # Add service account for cloud run
-        oidc_scheduler_hook(
-            token=SCHEDULER_OIDC_TOKEN,
+        oidc_scheduled_hook(
+            token=SCHEDULED_OIDC_TOKEN,
         ),
         # Wait for half an hour
-        deadline_scheduler_hook(duration=duration_pb2.Duration(seconds=1800)),
+        deadline_scheduled_hook(duration=duration_pb2.Duration(seconds=1800)),
     ),
 )
 
-router = APIRouter(route_class=TaskRoute, prefix="/tasks")
+task_router = APIRouter(route_class=DelayedRoute, prefix="/tasks")
 
 
-@router.post("/hello")
+@task_router.post("/hello")
 async def hello(p: Payload = Payload(message="Default")):
     message = f"Hello task ran with payload: {p.message}"
     logger.warning(message)
     return {"message": message}
 
 
-@router.post("/timed_hello")
+scheduled_router = APIRouter(route_class=ScheduledRoute, prefix="/scheduled")
+
+
+@scheduled_router.post("/timed_hello")
 async def scheduled_hello(p: Payload = Payload(message="Default")):
     message = f"Scheduled hello task ran with payload: {p.message}"
     logger.warning(message)
@@ -65,8 +75,9 @@ async def scheduled_hello(p: Payload = Payload(message="Default")):
 
 scheduled_hello.scheduler(
     name="testing-examples-scheduled-hello",
-    schedule="*/5 * * * *",
+    schedule="* * * * *",
     time_zone="Asia/Kolkata",
 ).schedule(p=Payload(message="Scheduled"))
 
-app.include_router(router)
+app.include_router(task_router)
+app.include_router(scheduled_router)

examples/simple/main.py

@@ -10,10 +10,10 @@
 from pydantic import BaseModel
 
 # Imports from this repository
-from fastapi_cloud_tasks.taskroute import TaskRouteBuilder
+from fastapi_cloud_tasks.delayed_route import DelayedRouteBuilder
 from fastapi_cloud_tasks.utils import queue_path
 
-TaskRoute = TaskRouteBuilder(
+DelayedRoute = DelayedRouteBuilder(
     # Base URL where the task server will get hosted
     base_url=os.getenv("TASK_LISTENER_BASE_URL", default="https://d860-35-208-83-220.ngrok.io"),
     # Full queue path to which we'll send tasks.
@@ -25,7 +25,7 @@
     ),
 )
 
-task_router = APIRouter(route_class=TaskRoute, prefix="/tasks")
+task_router = APIRouter(route_class=DelayedRoute, prefix="/tasks")
 
 logger = logging.getLogger("uvicorn")
 

fastapi_cloud_tasks/taskroute.py fastapi_cloud_tasks/delayed_route.pyfastapi_cloud_tasks/taskroute.py renamed to fastapi_cloud_tasks/delayed_route.py

@@ -8,46 +8,31 @@
 
 # Imports from this repository
 from fastapi_cloud_tasks.delayer import Delayer
-from fastapi_cloud_tasks.hooks import SchedulerHook
-from fastapi_cloud_tasks.hooks import TaskHook
+from fastapi_cloud_tasks.hooks import DelayedTaskHook
 from fastapi_cloud_tasks.hooks import noop_hook
 from fastapi_cloud_tasks.scheduler import Scheduler
 
 
-def TaskRouteBuilder(
+def DelayedRouteBuilder(
     *,
     base_url: str,
     queue_path: str,
     task_create_timeout: float = 10.0,
-    schedule_create_timeout: float = 10.0,
-    pre_create_hook: TaskHook = None,
-    pre_scheduler_hook: SchedulerHook = None,
+    pre_create_hook: DelayedTaskHook = None,
     client=None,
     scheduler_client=None,
 ):
     if client is None:
         client = tasks_v2.CloudTasksClient()
 
-    if scheduler_client is None:
-        scheduler_client = scheduler_v1.CloudSchedulerClient()
-
     if pre_create_hook is None:
         pre_create_hook = noop_hook
 
-    if pre_scheduler_hook is None:
-        pre_scheduler_hook = noop_hook
-
-    q_path = client.parse_queue_path(queue_path)
-    default_location_path = scheduler_client.common_location_path(
-        project=q_path["project"], location=q_path["location"]
-    )
-
     class TaskRouteMixin(APIRoute):
         def get_route_handler(self) -> Callable:
             original_route_handler = super().get_route_handler()
             self.endpoint.options = self.delayOptions
             self.endpoint.delay = self.delay
-            self.endpoint.scheduler = self.schedulerOptions
             return original_route_handler
 
         def delayOptions(self, **options) -> Delayer:
@@ -70,19 +55,4 @@ def delayOptions(self, **options) -> Delayer:
         def delay(self, **kwargs):
             return self.delayOptions().delay(**kwargs)
 
-        def schedulerOptions(self, *, name, schedule, **options) -> Scheduler:
-            schedulerOpts = dict(
-                base_url=base_url,
-                location_path=default_location_path,
-                client=scheduler_client,
-                pre_scheduler_hook=pre_scheduler_hook,
-                schedule_create_timeout=schedule_create_timeout,
-                name=name,
-                schedule=schedule,
-            )
-
-            schedulerOpts.update(options)
-
-            return Scheduler(route=self, **schedulerOpts)
-
     return TaskRouteMixin