Skip to content

Instantly share code, notes, and snippets.

@carlos-a-g-h
Last active May 20, 2023 05:06
Show Gist options
  • Save carlos-a-g-h/fbfbc8cb0634d608fda5861738024f63 to your computer and use it in GitHub Desktop.
Save carlos-a-g-h/fbfbc8cb0634d608fda5861738024f63 to your computer and use it in GitHub Desktop.
A small module for interacting with existing cron jobs using the cron-job.org API
#!/usr/bin/python3.9
import asyncio
import aiohttp
async def cronjob_patch(session:aiohttp.ClientSession,cronjob_apikey:str,cronjob_jid:str,cronjob_data:dict)->bool:
ts=False
if not session:
ts=True
session=aiohttp.ClientSession()
the_url=f"https://api.cron-job.org/jobs/{cronjob_jid}"
the_headers={"Accept":"*/*","Authorization":f"Bearer {cronjob_apikey}","Content-Type":"application/json"}
result=True
print("\n- PATCH:",the_url)
print(" HEADERS:",the_headers)
try:
async with session.request(method="PATCH",url=the_url,headers=the_headers,json=cronjob_data) as response:
print(" RESULT:",await response.text())
except Exception as e:
print(" ERROR:",e)
result=False
if ts:
await session.close()
return result
async def cronjob_off(session:aiohttp.ClientSession,cronjob_apikey:str,cronjob_jid:str)->bool:
ts=False
if not session:
ts=True
session=aiohttp.ClientSession()
result=await cronjob_patch(session,cronjob_apikey,cronjob_jid,{"job":{"enabled":False}})
if ts:
await session.close()
return result
async def cronjob_force(session:aiohttp.ClientSession,cronjob_apikey:str,cronjob_jid:str,target_url:str)->bool:
ts=False
if not session:
ts=True
session=aiohttp.ClientSession()
result=await cronjob_patch(session,cronjob_apikey,cronjob_jid,{"job":{"enabled":True,"url":target_url}})
if ts:
await session.close()
return result
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment