### Install AioClock

Source: https://github.com/manimozaffar/aioclock/blob/main/docs/index.md

Install the aioclock library using pip.

```bash
pip install aioclock
```

--------------------------------

### Full aioclock Example

Source: https://github.com/manimozaffar/aioclock/blob/main/README.md

This example demonstrates various features of aioclock including task scheduling with different triggers (Every, At, Forever, Once), dependency injection, task grouping, and application lifecycle management.

```python
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
import asyncio

from aioclock import AioClock, At, Depends, Every, Forever, Once
from aioclock.group import Group

# groups.py
group = Group()


def more_useless_than_me():
    return "I'm a dependency. I'm more useless than a screen door on a submarine."


@group.task(trigger=Every(seconds=10))
async def every():
    print("Every 10 seconds, I make a quantum leap. Where will I land next?")


@group.task(trigger=Every(seconds=5))
def even_sync_works():
    print("I'm a synchronous task. I work even in async world.")


@group.task(trigger=At(tz="UTC", hour=0, minute=0, second=0))
async def at():
    print(
        "When the clock strikes midnight... I turn into a pumpkin. Just kidding, I run this task!"
    )


@group.task(trigger=Forever())
async def forever(val: str = Depends(more_useless_than_me)):
    await asyncio.sleep(2)
    print("Heartbeat detected. Still not a zombie. Will check again in a bit.")
    assert val == "I'm a dependency. I'm more useless than a screen door on a submarine."


@group.task(trigger=Once())
async def once():
    print("Just once, I get to say something. Here it goes... I love lamp.")


@asynccontextmanager
async def lifespan(aio_clock: AioClock) -> AsyncGenerator[AioClock]:
    # starting up
    print(
        "Welcome to the Async Chronicles! Did you know a group of unicorns is called a blessing? Well, now you do!"
    )
    yield aio_clock
    # shuting down
    print("Going offline. Remember, if your code is running, you better go catch it!")


# app.py
app = AioClock(lifespan=lifespan)
app.include_group(group)


# main.py
if __name__ == "__main__":
    asyncio.run(app.serve())

```

--------------------------------

### Run AioClock with FastAPI Lifespan

Source: https://github.com/manimozaffar/aioclock/blob/main/docs/examples/fastapi.md

Integrate AioClock into FastAPI by starting its serve task during application startup and cancelling it on shutdown. This setup is generally not recommended for production environments.

```python
from aioclock import AioClock
from fastapi import FastAPI
import asyncio
from contextlib import asynccontextmanager

clock_app = AioClock()

@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(clock_app.serve())
    yield

    try:
        task.cancel()
        await task
    except asyncio.CancelledError:
        ...
```

--------------------------------

### AioClock Broker Integration Example

Source: https://github.com/manimozaffar/aioclock/blob/main/docs/examples/brokers.md

This snippet demonstrates setting up AioClock to manage a Redis broker. It includes defining a broker type, a singleton Redis instance, a lifespan context manager for connection handling, and a task to read messages from a queue. Use this pattern when you need AioClock to manage the lifecycle of your broker connections and provide them to tasks.

```python
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from aioclock import AioClock, Forever, Depends
from functools import lru_cache
from typing import NewType

BrokerType = NewType("BrokerType", ...) # your broker type ...

# your singleton redis instance
@lru_cache
def get_redis():
    ... # Replace with actual Redis client initialization

@asynccontextmanager
async def lifespan(aio_clock: AioClock, redis: BrokerType = Depends(get_redis)) -> AsyncGenerator[AioClock]:
    yield aio_clock
    await redis.disconnect() # Ensure disconnection on shutdown


app = AioClock(lifespan=lifespan)


@app.task(trigger=Forever())
async def read_message_queue(redis: BrokerType = Depends(get_redis)):
    async for message in redis.listen("..."): # Replace "..." with your queue name
        ... # Process the message
```

--------------------------------

### Implement Custom ExponentialBackoff Trigger

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Subclass BaseTrigger to create a custom trigger that doubles the wait time after each execution. This example demonstrates how to implement `should_trigger`, `get_waiting_time_till_next_trigger`, and `trigger_next` for an exponential backoff strategy.

```python
import asyncio
from typing import Literal
from aioclock.triggers import BaseTrigger
from aioclock import AioClock

class ExponentialBackoff(BaseTrigger[Literal["ExponentialBackoff"]]):
    """Doubles the wait time after each execution, starting from `initial_seconds`."""
    type_: Literal["ExponentialBackoff"] = "ExponentialBackoff"
    initial_seconds: float = 1.0
    _current_wait: float = 0.0

    def should_trigger(self) -> bool:
        return True  # run forever

    async def get_waiting_time_till_next_trigger(self):
        return self._current_wait

    async def trigger_next(self) -> None:
        wait = self._current_wait if self._current_wait > 0 else self.initial_seconds
        self._current_wait = wait * 2  # double next time
        await asyncio.sleep(wait)

app = AioClock()

@app.task(trigger=ExponentialBackoff(initial_seconds=2.0))
async def retry_with_backoff():
    print("Attempt — waits 2s, then 4s, then 8s, ...")

if __name__ == "__main__":
    asyncio.run(app.serve())

```

--------------------------------

### Schedule a Task for Single Execution with Once Trigger

Source: https://context7.com/manimozaffar/aioclock/llms.txt

The `Once` trigger ensures a task runs exactly one time, useful for one-off initializations after the application starts. It's equivalent to setting `max_loop_count=1`.

```python
from aioclock import AioClock, Once

app = AioClock()

@app.task(trigger=Once())
async def seed_database():
    print("Seeding database with initial data — runs once")
    await db.execute("INSERT INTO config VALUES (...)")

@app.task(trigger=Once())
async def warm_cache():
    print("Warming up application cache — runs once")
    await cache.preload()
```

--------------------------------

### Aioclock Application Entry Point

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Demonstrates setting up the main AioClock application, registering tasks with different triggers (Once, Every, Forever), and configuring lifespan hooks for startup and shutdown.

```python
import asyncio
from contextlib import asynccontextmanager
from aioclock import AioClock, Every, Once, Forever, Depends

# --- Lifespan: runs code before and after serve() ---
@asynccontextmanager
async def lifespan(app: AioClock):
    print("Starting up: connect to DB, load models, etc.")
    yield app
    print("Shutting down: release resources.")

app = AioClock(lifespan=lifespan)

# --- Register tasks directly on app ---
@app.task(trigger=Once())
async def initialize():
    print("Runs exactly once at startup")

@app.task(trigger=Every(seconds=30))
async def heartbeat():
    print("Runs every 30 seconds")

@app.task(trigger=Forever())
async def poll():
    await asyncio.sleep(1)
    print("Runs in a tight loop, re-entered immediately after each execution")

if __name__ == "__main__":
    asyncio.run(app.serve())
```

--------------------------------

### Task Scheduling with Dependency Injection

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Demonstrates scheduling tasks with dependencies and overriding them for testing.

```python
import asyncio
from aioclock import AioClock, Every
from aioclock.group import Group


group = Group()

@group.task(trigger=Every(seconds=60))
async def sync_records(
    db=Depends(get_db_connection),
    config=Depends(get_config)
):
    print(f"Syncing with db={db}, retry_limit={config['retry_limit']}")

# --- Override dependency for testing ---
def get_mock_db():
    return {"host": "mock", "db": "test_db"}

app = AioClock()
app.include_group(group)
app.override_dependencies(get_db_connection, get_mock_db)

if __name__ == "__main__":
    asyncio.run(app.serve())
```

--------------------------------

### Define and Schedule Tasks with AioClock

Source: https://github.com/manimozaffar/aioclock/blob/main/docs/index.md

Demonstrates defining tasks using groups and various triggers (Every, At, Forever, Once). Includes dependency injection and a lifespan context manager for the AioClock application. Synchronous tasks are also supported.

```python
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
import asyncio

from aioclock import AioClock, At, Depends, Every, Forever, Once
from aioclock.group import Group

# groups.py
group = Group()


def more_useless_than_me():
    return "I'm a dependency. I'm more useless than a screen door on a submarine."


@group.task(trigger=Every(seconds=10))
async def every():
    print("Every 10 seconds, I make a quantum leap. Where will I land next?")


@group.task(trigger=Every(seconds=5))
def even_sync_works():
    print("I'm a synchronous task. I work even in async world.")


@group.task(trigger=At(tz="UTC", hour=0, minute=0, second=0))
async def at():
    print(
        "When the clock strikes midnight... I turn into a pumpkin. Just kidding, I run this task!"
    )


@group.task(trigger=Forever())
async def forever(val: str = Depends(more_useless_than_me)):
    await asyncio.sleep(2)
    print("Heartbeat detected. Still not a zombie. Will check again in a bit.")
    assert val == "I'm a dependency. I'm more useless than a screen door on a submarine."


@group.task(trigger=Once())
async def once():
    print("Just once, I get to say something. Here it goes... I love lamp.")



@asynccontextmanager
async def lifespan(aio_clock: AioClock) -> AsyncGenerator[AioClock]:
    print(
        "Welcome to the Async Chronicles! Did you know a group of unicorns is called a blessing? Well, now you do!"
    )
    yield aio_clock
    print("Going offline. Remember, if your code is running, you better go catch it!")


app = AioClock(lifespan=lifespan)
app.include_group(group)

# main.py
if __name__ == "__main__":
    asyncio.run(app.serve())

```

--------------------------------

### Task Timeout Configuration

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Illustrates setting task timeouts at both the task and group levels. Tasks exceeding their timeout are cancelled and logged.

```python
import asyncio
from aioclock import AioClock, Every
from aioclock.group import Group

app = AioClock()

@app.task(trigger=Every(seconds=10), timeout=5.0)
async def must_finish_quickly():
    # If this takes more than 5 seconds, TaskTimeoutError is logged
    # and the task loop continues on next cycle
    await asyncio.sleep(3)  # OK — within 5s timeout
    print("Done")

# Group-level timeout: 8 seconds for all tasks, unless overridden
slow_group = Group(timeout=8)

@slow_group.task(trigger=Every(minutes=1))
async def task_with_group_timeout():
    await asyncio.sleep(6)  # OK — within 8s group timeout

@slow_group.task(trigger=Every(minutes=1), timeout=2)
async def task_with_task_timeout():
    await asyncio.sleep(3)  # Will timeout — TaskTimeoutError logged

app.include_group(slow_group)
```

--------------------------------

### Schedule Task at Specific Time of Day with At Trigger

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Use the `At` trigger to schedule tasks at a specific hour, minute, and second within a given timezone. You can also specify days of the week or keep it daily. Timezones are validated at startup.

```python
from aioclock import AioClock, At

app = AioClock()

@app.task(trigger=At(tz="UTC", hour=0, minute=0, second=0))
async def midnight_utc():
    print("Runs every day at midnight UTC")

@app.task(trigger=At(tz="America/New_York", hour=9, minute=30))
async def morning_report():
    print("Runs every day at 9:30 AM Eastern")

@app.task(trigger=At(tz="Europe/London", hour=17, minute=0, at="every friday"))
async def weekly_summary():
    print("Runs every Friday at 5 PM London time")

@app.task(trigger=At(tz="Asia/Kolkata", hour=12, minute=0, max_loop_count=3))
async def limited_noon_task():
    print("Runs at noon IST, but only 3 times total")
```

--------------------------------

### Aioclock Group for Modular Task Organization

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Shows how to use the `Group` class to organize tasks modularly, similar to routers in web frameworks. Tasks can be defined in separate modules and included into the main app. Group-level timeouts can be set and overridden per task.

```python
import asyncio
from aioclock import AioClock, Every, Once
from aioclock.group import Group

# --- email_tasks.py ---
email_group = Group(timeout=10)  # default 10s timeout for all tasks in group

@email_group.task(trigger=Every(minutes=5))
async def send_digest_emails():
    print("Sending digest emails...")

@email_group.task(trigger=Once(), timeout=30)  # overrides group timeout to 30s
async def send_welcome_email():
    print("Sending welcome email on startup")

# --- notification_tasks.py ---
notif_group = Group()

@notif_group.task(trigger=Every(hours=1))
async def push_notifications():
    print("Pushing notifications...")

# --- app.py ---
app = AioClock()
app.include_group(email_group)
app.include_group(notif_group)

if __name__ == "__main__":
    asyncio.run(app.serve())
```

--------------------------------

### Programmatic Task Inspection and Invocation

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Shows how to inspect task metadata and trigger tasks programmatically using the aioclock.api module. Dependencies are resolved automatically when invoking decorated functions.

```python
import asyncio
from aioclock import AioClock, Once, Every, Depends
from aioclock.api import get_metadata_of_all_tasks, run_specific_task, run_with_injected_deps

app = AioClock()

def provide_value():
    return 42

@app.task(trigger=Every(seconds=30))
async def periodic_job(val: int = Depends(provide_value)):
    print(f"Periodic job running with val={val}")
    return val

async def inspect_and_invoke():
    # List all registered tasks and their metadata
    tasks = await get_metadata_of_all_tasks(app)
    for t in tasks:
        print(f"Task: {t.task_name}, ID: {t.id}, Trigger: {t.trigger}, Next run: {t.trigger.expected_trigger_time}")

    # Run a specific task by its UUID immediately
    target_id = tasks[0].id
    await run_specific_task(target_id, app)

    # Run a decorated function directly with DI resolved
    result = await run_with_injected_deps(periodic_job)
    assert result == 42

asyncio.run(inspect_and_invoke())
```

--------------------------------

### Aioclock Trigger: Every - Interval-Based Scheduling

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Illustrates the `Every` trigger for interval-based scheduling. It supports various time units and configurations like `first_run_strategy` (immediate or wait) and `max_loop_count` for limiting executions.

```python
from aioclock import AioClock, Every

app = AioClock()

@app.task(trigger=Every(seconds=10))
def every_10_seconds():
    print("Runs every 10 seconds, waits 10s before first run")

@app.task(trigger=Every(minutes=2, first_run_strategy="immediate"))
def every_2_min_immediate():
    print("Runs immediately, then every 2 minutes")

@app.task(trigger=Every(hours=1, max_loop_count=5))
def hourly_limited():
    print("Runs every hour, stops after 5 executions")

@app.task(trigger=Every(days=1, hours=6))
def every_30_hours():
    print("Runs every 30 hours (1 day + 6 hours)")
```

--------------------------------

### Create a Continuous Loop Task with Forever Trigger

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Use the `Forever` trigger for tasks that should run continuously immediately after completion. Control the execution rate using `asyncio.sleep` within the task. Exceptions are caught and logged, allowing the loop to continue.

```python
from aioclock import AioClock, Forever
import asyncio

app = AioClock()

@app.task(trigger=Forever())
async def poll_queue():
    # Runs immediately, re-enters right after completion
    item = await some_queue.get()
    if item:
        await process(item)
    else:
        await asyncio.sleep(0.5)  # back-off when queue is empty

@app.task(trigger=Forever())
async def watchdog():
    await asyncio.sleep(5)
    health = await check_health()
    if not health:
        await alert_on_call()
```

--------------------------------

### Dependency Injection with Depends in Aioclock

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Aioclock integrates with `fast-depends` for dependency injection. Declare dependencies as function parameters using `Depends(callable)`. Dependencies can be sync or async, and implementations can be swapped using `app.override_dependencies()`.

```python
import asyncio
from aioclock import AioClock, Depends, Every
from aioclock.group import Group

# --- Dependencies ---
async def get_db_connection():
    # Simulates obtaining a DB connection
    return {"host": "localhost", "db": "mydb"}

def get_config():
    return {"retry_limit": 3}

```

--------------------------------

### FastAPI Extension for HTTP Control Plane

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Integrates aioclock with FastAPI, providing HTTP endpoints for task management. The aioclock scheduler runs as a background task within FastAPI's lifespan.

```python
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
import uvicorn

from aioclock import AioClock, Every
from aioclock.ext.fast import make_fastapi_router

clock_app = AioClock()

@clock_app.task(trigger=Every(seconds=3600))
async def hourly_job():
    print("Hourly job running...")

@clock_app.task(trigger=Every(minutes=10))
async def ten_min_job():
    print("10-minute job running...")

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start aioclock in the background alongside FastAPI
    task = asyncio.create_task(clock_app.serve())
    yield
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

app = FastAPI(lifespan=lifespan)
app.include_router(make_fastapi_router(clock_app))

# GET  /tasks              → list all tasks with metadata
# POST /task/{task_id}     → trigger task immediately (returns 404 if not found)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

--------------------------------

### Schedule Task with Cron Expression using Cron Trigger

Source: https://context7.com/manimozaffar/aioclock/llms.txt

The `Cron` trigger allows scheduling tasks using standard cron expressions (5 fields) and a timezone. It leverages `croniter` for parsing, supporting complex schedules.

```python
from aioclock import AioClock, Cron

app = AioClock()

@app.task(trigger=Cron(cron="0 12 * * *", tz="UTC"))
async def daily_noon():
    print("Every day at 12:00 UTC")

@app.task(trigger=Cron(cron="*/15 * * * *", tz="America/Chicago"))
async def every_15_min():
    print("Every 15 minutes")

@app.task(trigger=Cron(cron="0 9 * * 1-5", tz="Europe/Berlin"))
async def weekday_mornings():
    print("9 AM Mon-Fri, Berlin time")

@app.task(trigger=Cron(cron="0 0 1 * *", tz="UTC", max_loop_count=12))
async def monthly_first_12_months():
    print("First day of each month, stops after 12 runs")
```

--------------------------------

### Combine Multiple Triggers with OrTrigger

Source: https://context7.com/manimozaffar/aioclock/llms.txt

Use `OrTrigger` to create a task that fires when any of the provided inner triggers activate. Each inner trigger manages its own state, and the combined trigger stops when all inner triggers are exhausted.

```python
from aioclock import AioClock
from aioclock.triggers import OrTrigger, Every, At

app = AioClock()

@app.task(trigger=OrTrigger(
    triggers=[
        Every(seconds=30),                             # fires every 30 seconds
        At(tz="UTC", hour=8, minute=0, at="every day") # also fires daily at 8 AM UTC
    ]
))
async def flexible_task():
    print("Fires every 30s AND also at 8 AM UTC each day")

@app.task(trigger=OrTrigger(
    triggers=[
        Every(seconds=10, max_loop_count=5),   # fires 5 times
        Every(minutes=1, max_loop_count=3),    # fires 3 times
    ]
))
async def bounded_or_task():
    print("Total: 8 executions, then stops")
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.