<div align="center">
# aioskd
Asynchronous Background Task Scheduler
[](https://pypi.org/project/aioskd/)
[](https://codecov.io/gh/kradt/aioskd)
[](https://github.com/kradt/aioskd/actions/workflows/ci_tests.yml)
</div>
## Project Overview
aioskd is a powerful Python library designed to handle the execution of background tasks asynchronously and at scheduled intervals. It provides an efficient and flexible scheduler, making it effortless to integrate asynchronous background processing into your Python projects.
As applications grow more complex, certain tasks need to be executed in the background without affecting the responsiveness of the main application. AIOSKD is the perfect solution for such scenarios, as it allows you to offload these tasks to asynchronous workers, ensuring smooth execution and a better user experience.
## Features
- Asynchronous execution of background tasks
- Customizable scheduling of tasks at specified intervals
- Easy-to-use API for integrating with your Python projects
- Lightweight and efficient
## Installation
You can install `aioskd` using pip:
```bash
pip install aioskd
```
## Usage
### Creating the Scheduler
To get started, import the necessary modules and create an instance of the `Scheduler` class:
```python
from aioskd import Scheduler
skd = Scheduler()
```
## Scheduling Tasks
### Scheduling Tasks with Decorators (Existing Method)
You can schedule tasks using the schedule decorator as shown below:
```python
import datetime
import asyncio
@skd.schedule(interval=datetime.timedelta(seconds=1))
async def task_one():
print("Task One - Hello world!")
await asyncio.sleep(2) # Simulate some async work taking 2 seconds
@skd.schedule(interval=datetime.timedelta(seconds=5))
async def task_two():
print("Task Two - I'm running every 5 seconds!")
await asyncio.sleep(1) # Simulate some async work taking 1 second
```
### Scheduling Tasks without Decorators (New Method)
Alternatively, you can also register tasks without using decorators. Here's how you can do it:
```python
import datetime
import asyncio
from aioskd import Scheduler
skd = Scheduler()
async def test_task(name: str, age: int = 25):
print(f"Hello {name} with age {age}")
skd.register_task(test_task, "John", age=30).schedule(interval=datetime.timedelta(seconds=5))
skd.register_task(test_task, "Alice", age=28).schedule(interval=datetime.timedelta(seconds=2))
```
In this example, the test_task function is registered with the scheduler using the register_task method.
You can pass the function along with its arguments to register_task, and then schedule it with the desired interval using the schedule method.
### Running the Scheduler
To start the scheduler and run the scheduled tasks, you can use the `run()` method:
```python
skd.run()
```
### Command Line Usage
If you want to run the scheduled tasks from the command line, you can use the following command:
```bash
skd path/to/file/with/tasks:obj_of_skd
```
## Examples
### Example 1: Scheduling a Task to Fetch Data
```python
import datetime
import asyncio
import aiohttp
from aioskd import Scheduler
skd = Scheduler()
@skd.schedule(interval=datetime.timedelta(minutes=30))
async def fetch_data():
url = "https://api.example.com/data"
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
# Process and store the data as needed
print("Data fetched successfully!")
else:
print("Failed to fetch data. Status code:", response.status)
if __name__ == "__main__":
skd.run()
```
In this example, the `fetch_data` task is scheduled to run every 30 minutes. It sends a request to an API to fetch data and then processes the response accordingly.
### Example 2: Sending Scheduled Reminder Emails
```python
import datetime
import asyncio
import aiosmtplib
from email.message import EmailMessage
from aioskd import Scheduler
skd = Scheduler()
@skd.schedule(interval=datetime.timedelta(hours=24))
async def send_reminder_email():
email_content = "Hello! Just a friendly reminder that your appointment is tomorrow."
msg = EmailMessage()
msg.set_content(email_content)
msg["Subject"] = "Appointment Reminder"
msg["From"] = "your_email@example.com"
msg["To"] = "recipient@example.com"
async with aiosmtplib.SMTP("smtp.example.com", 587) as server:
await server.starttls()
await server.login("your_email@example.com", "your_email_password")
await server.send_message(msg)
if __name__ == "__main__":
skd.run()
```
This example schedules the send_reminder_email task to run once every 24 hours, sending a reminder email to a specified recipient about an upcoming appointment.
### Example 3: Sending Weather Update Emails
```python
import datetime
import asyncio
import aiohttp
import aiosmtplib
from email.message import EmailMessage
from aioskd import Scheduler
skd = Scheduler()
async def get_weather_data(city: str) -> dict:
url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=YOUR_OPENWEATHERMAP_API_KEY"
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
data = await response.json()
return data
async def send_weather_email(city: str, recipient: str):
data = await get_weather_data(city)
temperature = data["main"]["temp"]
description = data["weather"][0]["description"]
email_content = f"Weather Update for {city}:\n\nTemperature: {temperature} °C\nDescription: {description}"
msg = EmailMessage()
msg.set_content(email_content)
msg["Subject"] = f"Weather Update for {city}"
msg["From"] = "your_email@example.com"
msg["To"] = recipient
async with aiosmtplib.SMTP("smtp.example.com", 587) as server:
await server.starttls()
await server.login("your_email@example.com", "your_email_password")
await server.send_message(msg)
print(f"Weather update email sent to {recipient}")
# Register tasks with different cities and recipients
task1 = skd.register_task(send_weather_email, "London", recipient="recipient1@example.com")
task1.schedule(interval=datetime.timedelta(hours=1))
task2 = skd.register_task(send_weather_email, "New York", recipient="recipient2@example.com")
task2.schedule(interval=datetime.timedelta(hours=2))
if __name__ == "__main__":
skd.run()
```
This code demonstrates the use of AIOSKD to send weather updates via email.
### `schedule` Decorator
- **interval**: `datetime.timedelta`
- The interval between which the asynchronous function should be executed.
- **repeat**: `bool`
- A flag that indicates whether the asynchronous function should be repeated or executed only once. - If set to `True`, the function will be scheduled to run repeatedly at the specified interval. - If set to `False`, the function will be executed only once.
- **immediate**: `bool`
- A flag that controls the first execution of the scheduled function. - If set to `True`, the scheduler will execute the function immediately when starting the task for the FIRST TIME, and subsequent executions will be based on the interval. - If set to `False`, the first execution will wait for the interval before running.
- **iter_count** (optional): `int`
- The necessary number of repeats. This parameter is applicable only when `repeat` is `True`. - It specifies the maximum number of times the function should be repeated. - If not provided, the function will continue to be repeated indefinitely until the scheduler is stopped or the coroutine is cancelled.
## Contributing
If you'd like to contribute to this project, follow these steps:
1. Fork the repository and create a new branch.
2. Make your changes and test them thoroughly.
3. Submit a pull request, explaining the changes you've made.
Raw data
{
"_id": null,
"home_page": "",
"name": "aioskd",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": "",
"keywords": "async,scheduling,background-tasks,asynchronous-programming,scheduler-library,python,task-scheduler,background-processing,concurrency,asyncio,timed-tasks,library,python3,interval-tasks,task-scheduling,asynchronous-background-tasks,background-scheduler,background-jobs,background-execution,task-execution,scheduling-tasks,async-jobs,async-tasks,job-scheduler,task-runner,task-manager",
"author": "Artem Sydorenko",
"author_email": "kradworkmail@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/b2/3c/0c94b216495323a6bec593fe018bc101917c8e8c455ca778e4270e567d4b/aioskd-0.0.7.tar.gz",
"platform": null,
"description": "<div align=\"center\">\n\t\n# aioskd \n\t\nAsynchronous Background Task Scheduler\n\n[](https://pypi.org/project/aioskd/)\n\n[](https://codecov.io/gh/kradt/aioskd)\n[](https://github.com/kradt/aioskd/actions/workflows/ci_tests.yml)\n</div>\n\n## Project Overview\n\naioskd is a powerful Python library designed to handle the execution of background tasks asynchronously and at scheduled intervals. It provides an efficient and flexible scheduler, making it effortless to integrate asynchronous background processing into your Python projects.\n\nAs applications grow more complex, certain tasks need to be executed in the background without affecting the responsiveness of the main application. AIOSKD is the perfect solution for such scenarios, as it allows you to offload these tasks to asynchronous workers, ensuring smooth execution and a better user experience.\n\n\n## Features\n\n- Asynchronous execution of background tasks\n- Customizable scheduling of tasks at specified intervals\n- Easy-to-use API for integrating with your Python projects\n- Lightweight and efficient\n\n## Installation\n\nYou can install `aioskd` using pip:\n\n```bash\npip install aioskd\n```\n\n## Usage\n\n### Creating the Scheduler\n\nTo get started, import the necessary modules and create an instance of the `Scheduler` class:\n\n```python\nfrom aioskd import Scheduler\n\n\nskd = Scheduler()\n```\n\n## Scheduling Tasks\n\n### Scheduling Tasks with Decorators (Existing Method)\nYou can schedule tasks using the schedule decorator as shown below:\n\n```python\n\nimport datetime\nimport asyncio\n\n@skd.schedule(interval=datetime.timedelta(seconds=1))\nasync def task_one():\n print(\"Task One - Hello world!\") \n await asyncio.sleep(2) # Simulate some async work taking 2 seconds \n\n@skd.schedule(interval=datetime.timedelta(seconds=5))\nasync def task_two():\n print(\"Task Two - I'm running every 5 seconds!\")\n await asyncio.sleep(1) # Simulate some async work taking 1 second\n```\n\n### Scheduling Tasks without Decorators (New Method)\nAlternatively, you can also register tasks without using decorators. Here's how you can do it:\n\n```python\n\nimport datetime\nimport asyncio\nfrom aioskd import Scheduler\n\nskd = Scheduler()\n\nasync def test_task(name: str, age: int = 25):\n print(f\"Hello {name} with age {age}\")\n\nskd.register_task(test_task, \"John\", age=30).schedule(interval=datetime.timedelta(seconds=5))\nskd.register_task(test_task, \"Alice\", age=28).schedule(interval=datetime.timedelta(seconds=2))\n```\nIn this example, the test_task function is registered with the scheduler using the register_task method. \nYou can pass the function along with its arguments to register_task, and then schedule it with the desired interval using the schedule method.\n\n### Running the Scheduler\n\nTo start the scheduler and run the scheduled tasks, you can use the `run()` method:\n\n```python\nskd.run()\n```\n\n### Command Line Usage\n\nIf you want to run the scheduled tasks from the command line, you can use the following command:\n\n```bash\nskd path/to/file/with/tasks:obj_of_skd\n```\n\n## Examples\n\n### Example 1: Scheduling a Task to Fetch Data\n\n```python\nimport datetime\nimport asyncio\nimport aiohttp\nfrom aioskd import Scheduler\n\n\nskd = Scheduler()\n\n@skd.schedule(interval=datetime.timedelta(minutes=30))\nasync def fetch_data():\n url = \"https://api.example.com/data\"\n async with aiohttp.ClientSession() as session:\n async with session.get(url) as response:\n if response.status == 200:\n data = await response.json() \n\t\t# Process and store the data as needed \n\t\tprint(\"Data fetched successfully!\") \n else:\n\t\tprint(\"Failed to fetch data. Status code:\", response.status)\n\n\nif __name__ == \"__main__\":\n skd.run()\n```\n\t\nIn this example, the `fetch_data` task is scheduled to run every 30 minutes. It sends a request to an API to fetch data and then processes the response accordingly.\n\n### Example 2: Sending Scheduled Reminder Emails\n\n```python\nimport datetime\nimport asyncio\nimport aiosmtplib\nfrom email.message import EmailMessage\nfrom aioskd import Scheduler\n\n\nskd = Scheduler() \n\n@skd.schedule(interval=datetime.timedelta(hours=24))\nasync def send_reminder_email():\n email_content = \"Hello! Just a friendly reminder that your appointment is tomorrow.\"\n msg = EmailMessage()\n msg.set_content(email_content)\n msg[\"Subject\"] = \"Appointment Reminder\"\n msg[\"From\"] = \"your_email@example.com\"\n msg[\"To\"] = \"recipient@example.com\" \n async with aiosmtplib.SMTP(\"smtp.example.com\", 587) as server:\n\tawait server.starttls()\n\tawait server.login(\"your_email@example.com\", \"your_email_password\")\n\tawait server.send_message(msg)\n\nif __name__ == \"__main__\":\n skd.run()\n```\nThis example schedules the send_reminder_email task to run once every 24 hours, sending a reminder email to a specified recipient about an upcoming appointment.\n\n### Example 3: Sending Weather Update Emails\n\n```python\nimport datetime\nimport asyncio\nimport aiohttp\nimport aiosmtplib\nfrom email.message import EmailMessage\nfrom aioskd import Scheduler\n\n\nskd = Scheduler()\n\nasync def get_weather_data(city: str) -> dict:\n url = f\"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=YOUR_OPENWEATHERMAP_API_KEY\"\n async with aiohttp.ClientSession() as session:\n async with session.get(url) as response:\n data = await response.json()\n return data\n\n\nasync def send_weather_email(city: str, recipient: str):\n data = await get_weather_data(city)\n temperature = data[\"main\"][\"temp\"]\n description = data[\"weather\"][0][\"description\"]\n email_content = f\"Weather Update for {city}:\\n\\nTemperature: {temperature} \u00b0C\\nDescription: {description}\"\n msg = EmailMessage()\n msg.set_content(email_content)\n msg[\"Subject\"] = f\"Weather Update for {city}\"\n msg[\"From\"] = \"your_email@example.com\"\n msg[\"To\"] = recipient\n async with aiosmtplib.SMTP(\"smtp.example.com\", 587) as server:\n await server.starttls()\n await server.login(\"your_email@example.com\", \"your_email_password\")\n await server.send_message(msg)\n print(f\"Weather update email sent to {recipient}\") \n\n\n# Register tasks with different cities and recipients\ntask1 = skd.register_task(send_weather_email, \"London\", recipient=\"recipient1@example.com\")\ntask1.schedule(interval=datetime.timedelta(hours=1))\n\ntask2 = skd.register_task(send_weather_email, \"New York\", recipient=\"recipient2@example.com\")\ntask2.schedule(interval=datetime.timedelta(hours=2))\n\nif __name__ == \"__main__\":\n skd.run()\n\n```\nThis code demonstrates the use of AIOSKD to send weather updates via email.\n\n### `schedule` Decorator\n\n- **interval**: `datetime.timedelta` \n\t- The interval between which the asynchronous function should be executed. \n- **repeat**: `bool` \n\t- A flag that indicates whether the asynchronous function should be repeated or executed only once. - If set to `True`, the function will be scheduled to run repeatedly at the specified interval. - If set to `False`, the function will be executed only once. \n- **immediate**: `bool` \n\t- A flag that controls the first execution of the scheduled function. - If set to `True`, the scheduler will execute the function immediately when starting the task for the FIRST TIME, and subsequent executions will be based on the interval. - If set to `False`, the first execution will wait for the interval before running. \n- **iter_count** (optional): `int` \n\t- The necessary number of repeats. This parameter is applicable only when `repeat` is `True`. - It specifies the maximum number of times the function should be repeated. - If not provided, the function will continue to be repeated indefinitely until the scheduler is stopped or the coroutine is cancelled.\n\n## Contributing\n\nIf you'd like to contribute to this project, follow these steps:\n\n1. Fork the repository and create a new branch.\n2. Make your changes and test them thoroughly.\n3. Submit a pull request, explaining the changes you've made.\n",
"bugtrack_url": null,
"license": "",
"summary": "Tool for execute async background tasks",
"version": "0.0.7",
"project_urls": null,
"split_keywords": [
"async",
"scheduling",
"background-tasks",
"asynchronous-programming",
"scheduler-library",
"python",
"task-scheduler",
"background-processing",
"concurrency",
"asyncio",
"timed-tasks",
"library",
"python3",
"interval-tasks",
"task-scheduling",
"asynchronous-background-tasks",
"background-scheduler",
"background-jobs",
"background-execution",
"task-execution",
"scheduling-tasks",
"async-jobs",
"async-tasks",
"job-scheduler",
"task-runner",
"task-manager"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b23c0c94b216495323a6bec593fe018bc101917c8e8c455ca778e4270e567d4b",
"md5": "84e53e9e5f828687afcdf1daf01de5ab",
"sha256": "c836031a104a136f8fff700dcc76f8cd3a816eb2a668001f865df447806d5d66"
},
"downloads": -1,
"filename": "aioskd-0.0.7.tar.gz",
"has_sig": false,
"md5_digest": "84e53e9e5f828687afcdf1daf01de5ab",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 9702,
"upload_time": "2023-07-27T16:33:48",
"upload_time_iso_8601": "2023-07-27T16:33:48.624962Z",
"url": "https://files.pythonhosted.org/packages/b2/3c/0c94b216495323a6bec593fe018bc101917c8e8c455ca778e4270e567d4b/aioskd-0.0.7.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-07-27 16:33:48",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "aioskd"
}
JSON