### Control Crontab Start with Decorator Source: https://github.com/gawel/aiocron/blob/master/README.rst Initialize a crontab with `start=False` to prevent immediate execution upon definition. The `start()` method can then be called explicitly to begin the scheduling. The original function is accessible via the `.func` attribute. ```python @aiocron.crontab('1 9 * * 1-5', start=False) async def attime(): print('run') attime.start() asyncio.get_event_loop().run_forever() ``` -------------------------------- ### Manual Control of Scheduled Tasks Source: https://context7.com/gawel/aiocron/llms.txt Uses start=False to prevent automatic scheduling, allowing manual control via start() and stop() methods. ```python import aiocron import asyncio @aiocron.crontab('1 9 * * 1-5', start=False) # 9:01 AM on weekdays async def weekday_report(): print('Generating weekday report') return await generate_report() # Access the original function if needed original_func = weekday_report.func # Start scheduling when ready weekday_report.start() # Stop scheduling when needed weekday_report.stop() # Restart if required weekday_report.start() asyncio.get_event_loop().run_forever() ``` -------------------------------- ### Manually Instantiate Crontab Source: https://github.com/gawel/aiocron/blob/master/README.rst Instantiate a crontab object directly, providing the cron expression, the coroutine function, and the `start` parameter. This avoids using the decorator syntax. ```python cron = crontab('0 * * * *', func=yourcoroutine, start=False) ``` -------------------------------- ### Schedule Tasks with crontab Decorator Source: https://context7.com/gawel/aiocron/llms.txt Automatically schedules an async function to run based on a cron expression when the event loop starts. ```python import aiocron import asyncio @aiocron.crontab('*/30 * * * *') # Run every 30 minutes async def scheduled_task(): print('Task executed at scheduled time') # Perform your async work here await some_async_operation() # Start the event loop - tasks will run automatically asyncio.get_event_loop().run_forever() ``` -------------------------------- ### Await Crontab for Next Execution with Arguments Source: https://github.com/gawel/aiocron/blob/master/README.rst A crontab instance can be awaited to get the result of its next scheduled execution. The coroutine can accept arguments, and error handling for exceptions during execution is recommended. ```python @aiocron.crontab('0 9,10 * * * mon,fri', start=False) async def attime(i): print('run %i' % i) async def once(): try: res = await attime.next(1) except Exception as e: print('It failed (%r)' % e) else: print(res) asyncio.get_event_loop().run_forever() ``` -------------------------------- ### Use Crontab as a Sleep Coroutine Source: https://github.com/gawel/aiocron/blob/master/README.rst A crontab instance can be used to pause execution until the next scheduled time, similar to a sleep function. This example waits until the next hour. ```python await crontab('0 * * * *').next() ``` -------------------------------- ### Initialize Cron Objects Directly Source: https://context7.com/gawel/aiocron/llms.txt Creates a Cron object manually to pass specific functions, arguments, and loop configurations. ```python import aiocron from aiocron import crontab import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) async def my_task(user_id, message="Hello"): print(f"Processing task for user {user_id}: {message}") return {"user_id": user_id, "status": "completed"} # Create crontab with function and arguments cron_job = crontab( '0 * * * *', # Every hour func=my_task, args=(123,), # Positional arguments kwargs={"message": "Scheduled run"}, # Keyword arguments start=False, loop=loop ) # Start when ready cron_job.start() # Access job info print(str(cron_job)) # Shows: "0 * * * * " print(repr(cron_job)) # Shows: ">" loop.run_forever() ``` -------------------------------- ### Execute Commands via CLI Source: https://context7.com/gawel/aiocron/llms.txt Schedule shell commands directly from the command line using cron syntax. ```bash # Run a command every minute python -m aiocron "* * * * *" echo "Hello World" # Run a script every hour, limited to 5 executions python -m aiocron -n 5 "0 * * * *" python /path/to/script.py # Run indefinitely (n=0) python -m aiocron -n 0 "*/5 * * * *" /path/to/backup.sh # Help python -m aiocron --help ``` -------------------------------- ### Handle Task Exceptions Source: https://context7.com/gawel/aiocron/llms.txt Capture exceptions raised within cron tasks using the next() method or allow them to propagate in auto-start mode. ```python import aiocron from aiocron import crontab import asyncio loop = asyncio.new_event_loop() @crontab("* * * * *", start=False, loop=loop) async def risky_task(should_fail=False): if should_fail: raise ValueError("Task failed intentionally") return "Task completed successfully" async def main(): # Handle exceptions from next() try: result = await risky_task.next(should_fail=True) except ValueError as e: print(f"Caught expected error: {e}") # Successful execution try: result = await risky_task.next(should_fail=False) print(f"Success: {result}") except Exception as e: print(f"Unexpected error: {e}") loop.run_until_complete(main()) ``` -------------------------------- ### Await Scheduled Execution with next() Source: https://context7.com/gawel/aiocron/llms.txt Uses the next() method to await the next scheduled occurrence of a task, optionally passing arguments. ```python import aiocron from aiocron import crontab import asyncio @aiocron.crontab('0 9,10 * * * mon,fri', start=False) async def process_data(item_id): print(f'Processing item {item_id}') return {"item_id": item_id, "status": "processed"} async def main(): # Await next scheduled execution with arguments for i in range(3): try: result = await process_data.next(i) print(f'Result: {result}') except Exception as e: print(f'Processing failed: {repr(e)}') asyncio.get_event_loop().run_until_complete(main()) ``` -------------------------------- ### Schedule Tasks with Timezones Source: https://context7.com/gawel/aiocron/llms.txt Use the tz parameter with ZoneInfo to schedule tasks in specific timezones. ```python import aiocron from aiocron import crontab import asyncio from zoneinfo import ZoneInfo # Python 3.9+ loop = asyncio.new_event_loop() # Use specific timezone eastern_tz = ZoneInfo("America/New_York") @crontab('0 9 * * *', tz=eastern_tz, loop=loop) async def morning_eastern(): print('Good morning Eastern time!') # UTC timezone utc_tz = ZoneInfo("UTC") @crontab('0 14 * * *', tz=utc_tz, loop=loop) async def afternoon_utc(): print('Afternoon UTC task') loop.run_forever() ``` -------------------------------- ### Implement Sleep Timers with crontab Source: https://context7.com/gawel/aiocron/llms.txt Uses crontab without a function to pause execution until the next matching cron time. ```python import aiocron from aiocron import crontab import asyncio async def main(): # Wait until the next hour print("Waiting until next hour...") await crontab('0 * * * *').next() print("It's now the top of the hour!") # Wait until next minute print("Waiting until next minute...") result = await crontab('* * * * *').next(42) # Pass value through print(f"Minute passed, received: {result}") # Output: (42,) # Wait until specific time (e.g., 9 AM) await crontab('0 9 * * *').next() print("It's 9 AM!") asyncio.get_event_loop().run_until_complete(main()) ``` -------------------------------- ### Integrate aiocron with Threading Source: https://context7.com/gawel/aiocron/llms.txt Run scheduled tasks in a dedicated background thread while the main thread performs other operations. ```python import threading import asyncio import aiocron import time class CronThread(threading.Thread): def __init__(self): super(CronThread, self).__init__() self.loop = None self.start() time.sleep(0.1) # Give time for the loop to start def run(self): self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) self.loop.run_forever() def stop(self): if self.loop: self.loop.call_soon_threadsafe(self.loop.stop) self.join() self.loop.close() def crontab(self, *args, **kwargs): kwargs["loop"] = self.loop return aiocron.crontab(*args, **kwargs) # Create the cron thread cron = CronThread() @cron.crontab("* * * * * *") # Every second (6-field format) async def background_task(): await asyncio.sleep(0.1) print("Background task executed") # Main thread can do other work try: time.sleep(5) # Let it run for 5 seconds finally: cron.stop() print("Cron stopped") ``` -------------------------------- ### Define Extended Cron Expressions Source: https://context7.com/gawel/aiocron/llms.txt Utilize 5-field or 6-field (with seconds) cron expressions for flexible scheduling. ```python import aiocron from aiocron import crontab import asyncio loop = asyncio.new_event_loop() # Standard 5-field format: minute hour day month weekday @crontab('0 9 * * 1-5', loop=loop) # 9 AM on weekdays async def weekday_morning(): print("Weekday morning task") # Every 15 minutes @crontab('*/15 * * * *', loop=loop) async def quarter_hourly(): print("Running every 15 minutes") # Specific days and times @crontab('30 14 1,15 * *', loop=loop) # 2:30 PM on 1st and 15th async def bimonthly_task(): print("Bimonthly task") # 6-field format with seconds: second minute hour day month weekday @crontab('*/10 * * * * *', loop=loop) # Every 10 seconds async def frequent_task(): print("Running every 10 seconds") # Named days @crontab('0 12 * * mon,wed,fri', loop=loop) # Noon on Mon, Wed, Fri async def alternating_days(): print("Alternating days task") loop.run_forever() ``` -------------------------------- ### Run Function at Scheduled Time with Decorator Source: https://github.com/gawel/aiocron/blob/master/README.rst Use the @aiocron.crontab decorator to schedule an async function to run at specified cron intervals. The asyncio event loop must be running for the scheduled function to execute. ```python import aiocron import asyncio @aiocron.crontab('*/30 * * * *') async def attime(): print('run') asyncio.get_event_loop().run_forever() ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.