# Robit
Chronological Automation Service Framework
A robot for your bits!
(Pronounced "Row-Bit")
## Features
- Lightweight (only one installed dependencies pytz which is also lightweight)
- Can run with monitoring interface or headless
- Very simple and easy to set up and configure
### Interface
## Usage
### Worker
- Code below is provided in the examples/worker_example.py file of this project.
```python
import random
from time import sleep
import robit
robit.set_time_zone('America/Edmonton')
robit.set_database_logging(True)
def function_to_alert_me(**kwargs):
print(f"ALARM !!!! {kwargs['alert_message']}")
wo = robit.Worker(
name='Robit Example Worker',
key='Your-Own-Unique-Worker-Key-That-Is-Secure',
web_server=True,
# web_server_address='0.0.0.0',
# web_server_port=8000,
alert_method=function_to_alert_me,
alert_health_threshold=99.0,
)
def function_sleep_short():
sleep(2)
return 'Slept for 2 seconds'
def function_sleep_for_time(sleep_time: int):
sleep(sleep_time)
return f'Slept for {sleep_time} seconds'
wo.add_job(
'Specific Sleep Period Function',
function_sleep_for_time,
method_kwargs={'sleep_time': 5},
group='Sleeping',
cron='* * * * *',
)
wo.add_job(
'Sleep 3 Seconds Function Every 5 Seconds',
function_sleep_for_time,
method_kwargs={'sleep_time': 3},
group='Sleeping',
cron='*/5 * * * * *',
)
wo.add_job(
'Sleep for Short Period',
function_sleep_short,
group='Sleeping',
cron='*/2 * * * *',
)
def function_random_fail_often():
if 3 <= random.randint(1,4):
division_by_zero = 5 / 0
sleep(4)
return 'No Error'
def function_random_fail_rare():
if 1 == random.randint(1,20):
division_by_zero = 5 / 0
sleep(4)
return 'No Error'
wo.add_job(
'A Function that Fails',
function_random_fail_often,
group='Failing',
cron='*/10 * * * * *',
retry_attempts=4,
)
wo.add_job(
'Might Fail Some Times',
function_random_fail_rare,
group='Failing',
cron='* * * * *',
)
def function_full_speed():
x = int()
for i in range(99999999):
x = i * i
sleep(1)
return f'Max multiplication result of {x:,}'
wo.add_job(
'Lower Delay Function',
function_full_speed,
group='Rapid Execution',
cron='* * * * * *',
)
def function_send_worker_information(worker: robit.Worker):
return f'Worker Success Count: {worker.success_count}'
wo.add_job('Send Worker Information', function_send_worker_information, group='Webhooks Or Database Update', cron='*/30 * * * * *')
if __name__ == '__main__':
wo.start()
```
The server will start and host a web portal on default port 8100 locally for you to view what is going on.
## Other Libraries Used
- Boostrap (Responsive UI)
- Alpine (Better UX)
- PyTZ (Awesome Time Zone Management)
Raw data
{
"_id": null,
"home_page": "https://github.com/stratusadv/robit",
"name": "robit",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": "",
"keywords": "automation,bot,cron,cronjob,chronological,worker,job",
"author": "Nathan Johnson",
"author_email": "nathanj@stratusadv.com",
"download_url": "https://files.pythonhosted.org/packages/39/29/0f7d421d6aab560556bd0372d7a2e1fc162b7664044a938b444f09fef10b/robit-0.4.5.tar.gz",
"platform": null,
"description": "# Robit\n\nChronological Automation Service Framework\n\nA robot for your bits! \n\n(Pronounced \"Row-Bit\")\n\n## Features\n\n- Lightweight (only one installed dependencies pytz which is also lightweight)\n- Can run with monitoring interface or headless\n- Very simple and easy to set up and configure \n\n### Interface\n\n\n\n## Usage\n\n### Worker\n\n- Code below is provided in the examples/worker_example.py file of this project.\n\n```python\nimport random\nfrom time import sleep\n\nimport robit\n\n\nrobit.set_time_zone('America/Edmonton')\nrobit.set_database_logging(True)\n\n\ndef function_to_alert_me(**kwargs):\n print(f\"ALARM !!!! {kwargs['alert_message']}\")\n\n\nwo = robit.Worker(\n name='Robit Example Worker',\n key='Your-Own-Unique-Worker-Key-That-Is-Secure',\n web_server=True,\n # web_server_address='0.0.0.0',\n # web_server_port=8000,\n alert_method=function_to_alert_me,\n alert_health_threshold=99.0,\n)\n\n\ndef function_sleep_short():\n sleep(2)\n return 'Slept for 2 seconds'\n\n\ndef function_sleep_for_time(sleep_time: int):\n sleep(sleep_time)\n return f'Slept for {sleep_time} seconds'\n\n\nwo.add_job(\n 'Specific Sleep Period Function',\n function_sleep_for_time,\n method_kwargs={'sleep_time': 5},\n group='Sleeping',\n cron='* * * * *',\n)\n\nwo.add_job(\n 'Sleep 3 Seconds Function Every 5 Seconds',\n function_sleep_for_time,\n method_kwargs={'sleep_time': 3},\n group='Sleeping',\n cron='*/5 * * * * *',\n)\n\nwo.add_job(\n 'Sleep for Short Period',\n function_sleep_short,\n group='Sleeping',\n cron='*/2 * * * *',\n)\n\n\ndef function_random_fail_often():\n if 3 <= random.randint(1,4):\n division_by_zero = 5 / 0\n sleep(4)\n return 'No Error'\n\n\ndef function_random_fail_rare():\n if 1 == random.randint(1,20):\n division_by_zero = 5 / 0\n sleep(4)\n return 'No Error'\n\n\nwo.add_job(\n 'A Function that Fails',\n function_random_fail_often,\n group='Failing',\n cron='*/10 * * * * *',\n retry_attempts=4,\n)\n\nwo.add_job(\n 'Might Fail Some Times',\n function_random_fail_rare,\n group='Failing',\n cron='* * * * *',\n)\n\n\ndef function_full_speed():\n x = int()\n for i in range(99999999):\n x = i * i\n sleep(1)\n return f'Max multiplication result of {x:,}'\n\n\nwo.add_job(\n 'Lower Delay Function',\n function_full_speed,\n group='Rapid Execution',\n cron='* * * * * *',\n)\n\n\ndef function_send_worker_information(worker: robit.Worker):\n return f'Worker Success Count: {worker.success_count}'\n\n\nwo.add_job('Send Worker Information', function_send_worker_information, group='Webhooks Or Database Update', cron='*/30 * * * * *')\n\n\nif __name__ == '__main__':\n wo.start()\n```\n\nThe server will start and host a web portal on default port 8100 locally for you to view what is going on.\n\n\n## Other Libraries Used\n\n- Boostrap (Responsive UI)\n- Alpine (Better UX)\n- PyTZ (Awesome Time Zone Management)\n\n\n",
"bugtrack_url": null,
"license": "",
"summary": "Service Worker Framework",
"version": "0.4.5",
"project_urls": {
"Homepage": "https://github.com/stratusadv/robit"
},
"split_keywords": [
"automation",
"bot",
"cron",
"cronjob",
"chronological",
"worker",
"job"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "1440110c226e2177ce657c2f17943038e2a5ee6f773619d9c04eb30768e23577",
"md5": "110e7af0314629b62e93698e24f61b63",
"sha256": "77ee385fe6d436d89ba140d5cb8996ca9a6e2e02c9e39ae6bba2b5d9b3646dad"
},
"downloads": -1,
"filename": "robit-0.4.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "110e7af0314629b62e93698e24f61b63",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 105760,
"upload_time": "2024-01-26T22:19:27",
"upload_time_iso_8601": "2024-01-26T22:19:27.396422Z",
"url": "https://files.pythonhosted.org/packages/14/40/110c226e2177ce657c2f17943038e2a5ee6f773619d9c04eb30768e23577/robit-0.4.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "39290f7d421d6aab560556bd0372d7a2e1fc162b7664044a938b444f09fef10b",
"md5": "d493c2314657d68e1c8675b977be1d4f",
"sha256": "b908c9db9a57b231dacb0bafaeb0f8367b320bfa4afec51b9e1dd29a89b43307"
},
"downloads": -1,
"filename": "robit-0.4.5.tar.gz",
"has_sig": false,
"md5_digest": "d493c2314657d68e1c8675b977be1d4f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 93008,
"upload_time": "2024-01-26T22:19:30",
"upload_time_iso_8601": "2024-01-26T22:19:30.094737Z",
"url": "https://files.pythonhosted.org/packages/39/29/0f7d421d6aab560556bd0372d7a2e1fc162b7664044a938b444f09fef10b/robit-0.4.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-01-26 22:19:30",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "stratusadv",
"github_project": "robit",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "robit"
}
JSON
