# PyAsyncScheduler
## Description
This is an asynchronous scheduler designed to execute both synchronous and asynchronous tasks with flexible timing and configuration options. It supports cron expressions, sleep durations, fixed intervals, start/end time windows, and occurrence limits.
Tasks can be defined using simple wordlists, added dynamically at runtime, and configured individually with execution constraints. While most tasks are expected to be asynchronous, synchronous functions are fully supported.
Optional features include integration with a vault for secure configuration or secrets management. The scheduler is built with extensibility and clarity in mind, suitable for running background jobs, periodic checks, lightweight automation, or orchestrating custom workflows in asynchronous Python applications.
## Requirements
This package require:
- python3
- python3 Standard Library
### Optional
- UrlTasker
- MiniVault
- JsonRpcExtended
#### Optional Dependencies
> Not used in this package but it's required for multiples optional requirements.
- PegParser (in JsonRpcExtended and UrlTasker)
- RC6Encryption (in MiniVault)
## Installation
### Pip
```bash
python3 -m pip install PyAsyncScheduler
```
### Git
```bash
git clone "https://github.com/mauricelambert/PyAsyncScheduler.git"
cd "PyAsyncScheduler"
python3 -m pip install .
```
### Wget
```bash
wget https://github.com/mauricelambert/PyAsyncScheduler/archive/refs/heads/main.zip
unzip main.zip
cd PyAsyncScheduler-main
python3 -m pip install .
```
### cURL
```bash
curl -O https://github.com/mauricelambert/PyAsyncScheduler/archive/refs/heads/main.zip
unzip main.zip
cd PyAsyncScheduler-main
python3 -m pip install .
```
## Usages
### Command line
> Command line require all optional requirements.
```bash
PyAsyncScheduler # Using CLI package executable
python3 -m PyAsyncScheduler # Using python module
python3 PyAsyncScheduler.pyz # Using python executable
PyAsyncScheduler.exe # Using python Windows executable
PyAsyncScheduler -v myvault tasks1.json tasks2.json ... tasksN.json
```
### Python script
```python
from PyAsyncScheduler import *
async def my_async_template(template, **kwargs):
# kwargs are used in environment variables for subprocess tasks
print("Do something with template", template, kwargs)
scheduler = TaskScheduler(start_callable=lambda x, **y: my_async_template(x, **y))
scheduler.add_task({
"template": "https://api.com/foo?u=${user}&p=${pass}",
"csv_inputs": ["credentials.csv"],
"sleep": 3600,
})
run(scheduler.run())
```
```python
from MiniVault import PasswordVault
from PyAsyncScheduler import *
from functools import partial
from getpass import getpass
from asyncio import sleep
async def my_async_service():
# kwargs are used in environment variables for subprocess tasks
while True:
await sleep(1)
print("Do something in async loop")
def my_sync_template(url, **kwargs):
# kwargs are used in environment variables for subprocess tasks
print("Do something with template", template, kwargs)
def handle_task_completed(template, task):
print("Task complete:", template, task["id"], task)
PasswordVault.start(master_password=getpass(), root_dir="test")
scheduler = TaskScheduler(
start_callable=lambda x, **y: partial(my_sync_template, x, **y),
worker_count=8,
process_result=handle_task_completed,
external_coroutines=[my_async_service()],
vault=vault,
)
scheduler.add_task({
"template": "https://api.com/foo?u=${user}&p=${pass}",
"csv_inputs": ["users.csv", "passes.csv"],
"start_time": "2025-06-22T00:00:00",
"end_time": "2025-06-22T06:00:00",
"occurrences": 12,
})
scheduler.add_task({
"template": "https://${edr_user}:${edr_password}@edr.com/foo?u=${user}&p=${pass}",
"csv_inputs": ["users.csv", "passes.csv"],
"credentials": {"edr": {"category": "EDR", "role": "events_reader"}},
"cron": "*/5 * * * *",
"limit": {"max_executions": 100, "per_seconds": 60},
"instance_spacing": 1
})
run(scheduler.run())
```
## Links
- [Pypi](https://pypi.org/project/PyAsyncScheduler)
- [Github](https://github.com/mauricelambert/PyAsyncScheduler)
- [Documentation](https://mauricelambert.github.io/info/python/code/PyAsyncScheduler.html)
## License
Licensed under the [GPL, version 3](https://www.gnu.org/licenses/).
Raw data
{
"_id": null,
"home_page": "https://github.com/mauricelambert/PyAsyncScheduler",
"name": "PyAsyncScheduler",
"maintainer": "Maurice Lambert",
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "Maurice Lambert <mauricelambert434@gmail.com>",
"keywords": "scheduler, cron, async, task-scheduler, background-tasks, vault",
"author": "Maurice Lambert",
"author_email": "Maurice Lambert <mauricelambert434@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/3c/c8/eed002d69637af4b56b57e7ca77fec8ff1a8fe22de4060eb574f02abf90b/pyasyncscheduler-0.0.1.tar.gz",
"platform": "Windows",
"description": "![PyAsyncScheduler Logo](https://mauricelambert.github.io/info/python/code/PyAsyncScheduler_small.png \"PyAsyncScheduler logo\")\n\n# PyAsyncScheduler\n\n## Description\n\nThis is an asynchronous scheduler designed to execute both synchronous and asynchronous tasks with flexible timing and configuration options. It supports cron expressions, sleep durations, fixed intervals, start/end time windows, and occurrence limits.\n\nTasks can be defined using simple wordlists, added dynamically at runtime, and configured individually with execution constraints. While most tasks are expected to be asynchronous, synchronous functions are fully supported.\n\nOptional features include integration with a vault for secure configuration or secrets management. The scheduler is built with extensibility and clarity in mind, suitable for running background jobs, periodic checks, lightweight automation, or orchestrating custom workflows in asynchronous Python applications.\n\n## Requirements\n\nThis package require:\n\n - python3\n - python3 Standard Library\n\n### Optional\n\n - UrlTasker\n - MiniVault\n - JsonRpcExtended\n\n#### Optional Dependencies\n\n> Not used in this package but it's required for multiples optional requirements.\n\n - PegParser (in JsonRpcExtended and UrlTasker)\n - RC6Encryption (in MiniVault)\n\n## Installation\n\n### Pip\n\n```bash\npython3 -m pip install PyAsyncScheduler\n```\n\n### Git\n\n```bash\ngit clone \"https://github.com/mauricelambert/PyAsyncScheduler.git\"\ncd \"PyAsyncScheduler\"\npython3 -m pip install .\n```\n\n### Wget\n\n```bash\nwget https://github.com/mauricelambert/PyAsyncScheduler/archive/refs/heads/main.zip\nunzip main.zip\ncd PyAsyncScheduler-main\npython3 -m pip install .\n```\n\n### cURL\n\n```bash\ncurl -O https://github.com/mauricelambert/PyAsyncScheduler/archive/refs/heads/main.zip\nunzip main.zip\ncd PyAsyncScheduler-main\npython3 -m pip install .\n```\n\n## Usages\n\n### Command line\n\n> Command line require all optional requirements.\n\n```bash\nPyAsyncScheduler # Using CLI package executable\npython3 -m PyAsyncScheduler # Using python module\npython3 PyAsyncScheduler.pyz # Using python executable\nPyAsyncScheduler.exe # Using python Windows executable\n\nPyAsyncScheduler -v myvault tasks1.json tasks2.json ... tasksN.json\n```\n\n### Python script\n\n```python\nfrom PyAsyncScheduler import *\n\nasync def my_async_template(template, **kwargs):\n # kwargs are used in environment variables for subprocess tasks\n print(\"Do something with template\", template, kwargs)\n\nscheduler = TaskScheduler(start_callable=lambda x, **y: my_async_template(x, **y))\nscheduler.add_task({\n \"template\": \"https://api.com/foo?u=${user}&p=${pass}\",\n \"csv_inputs\": [\"credentials.csv\"],\n \"sleep\": 3600,\n})\nrun(scheduler.run())\n```\n\n```python\nfrom MiniVault import PasswordVault\nfrom PyAsyncScheduler import *\nfrom functools import partial\nfrom getpass import getpass\nfrom asyncio import sleep\n\nasync def my_async_service():\n # kwargs are used in environment variables for subprocess tasks\n while True:\n await sleep(1)\n print(\"Do something in async loop\")\n\ndef my_sync_template(url, **kwargs):\n # kwargs are used in environment variables for subprocess tasks\n print(\"Do something with template\", template, kwargs)\n\ndef handle_task_completed(template, task):\n print(\"Task complete:\", template, task[\"id\"], task)\n\nPasswordVault.start(master_password=getpass(), root_dir=\"test\")\n\nscheduler = TaskScheduler(\n start_callable=lambda x, **y: partial(my_sync_template, x, **y),\n worker_count=8,\n process_result=handle_task_completed,\n external_coroutines=[my_async_service()],\n vault=vault,\n)\nscheduler.add_task({\n \"template\": \"https://api.com/foo?u=${user}&p=${pass}\",\n \"csv_inputs\": [\"users.csv\", \"passes.csv\"],\n \"start_time\": \"2025-06-22T00:00:00\",\n \"end_time\": \"2025-06-22T06:00:00\",\n \"occurrences\": 12,\n})\nscheduler.add_task({\n \"template\": \"https://${edr_user}:${edr_password}@edr.com/foo?u=${user}&p=${pass}\",\n \"csv_inputs\": [\"users.csv\", \"passes.csv\"],\n \"credentials\": {\"edr\": {\"category\": \"EDR\", \"role\": \"events_reader\"}},\n \"cron\": \"*/5 * * * *\",\n \"limit\": {\"max_executions\": 100, \"per_seconds\": 60},\n \"instance_spacing\": 1\n})\nrun(scheduler.run())\n```\n\n## Links\n\n - [Pypi](https://pypi.org/project/PyAsyncScheduler)\n - [Github](https://github.com/mauricelambert/PyAsyncScheduler)\n - [Documentation](https://mauricelambert.github.io/info/python/code/PyAsyncScheduler.html)\n\n## License\n\nLicensed under the [GPL, version 3](https://www.gnu.org/licenses/).\n",
"bugtrack_url": null,
"license": "GPL-3.0 License",
"summary": "An asynchronous task scheduler, with cron syntax, intervals, limits, dynamic configuration, and optional vault integration.",
"version": "0.0.1",
"project_urls": {
"Documentation": "https://mauricelambert.github.io/info/python/code/PyAsyncScheduler.html",
"Github": "https://github.com/mauricelambert/PyAsyncScheduler",
"Homepage": "https://github.com/mauricelambert/PyAsyncScheduler"
},
"split_keywords": [
"scheduler",
" cron",
" async",
" task-scheduler",
" background-tasks",
" vault"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "3cc8eed002d69637af4b56b57e7ca77fec8ff1a8fe22de4060eb574f02abf90b",
"md5": "56da1b6ab06ecde4a936682300331571",
"sha256": "d2728f4e7f6d95104a490e9447720935668709bd1418a44280ad2202a80d58f9"
},
"downloads": -1,
"filename": "pyasyncscheduler-0.0.1.tar.gz",
"has_sig": false,
"md5_digest": "56da1b6ab06ecde4a936682300331571",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 25684,
"upload_time": "2025-09-01T05:03:42",
"upload_time_iso_8601": "2025-09-01T05:03:42.848285Z",
"url": "https://files.pythonhosted.org/packages/3c/c8/eed002d69637af4b56b57e7ca77fec8ff1a8fe22de4060eb574f02abf90b/pyasyncscheduler-0.0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-01 05:03:42",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "mauricelambert",
"github_project": "PyAsyncScheduler",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [
{
"name": "UrlTasker",
"specs": []
},
{
"name": "MiniVault",
"specs": []
},
{
"name": "JsonRpcExtended",
"specs": []
}
],
"lcname": "pyasyncscheduler"
}
JSON
