vassili/seitime-frappe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 4

style: Add type hints, Black-ish & f-strings

Browse source
This commit is contained in:
Gavin D'souza 2021-02-17 19:07:08 +05:30
parent 53378ac7e1
commit c8a0bd000d
1 changed files with 31 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
frappe/core/doctype/scheduled_job_type/scheduled_job_type.py
View file

@ -3,6 +3,7 @@
# For license information, please see license.txt
from __future__ import unicode_literals
from typing import Dict, List
import frappe, json
from frappe.model.document import Document
@ -11,12 +12,13 @@ from datetime import datetime
from croniter import croniter
from frappe.utils.background_jobs import enqueue, get_jobs
class ScheduledJobType(Document):
def autoname(self):
self.name = '.'.join(self.method.split('.')[-2:])
self.name = ".".join(self.method.split(".")[-2:])
def validate(self):
if self.frequency != 'All':
if self.frequency != "All":
# force logging for all events other than continuous ones (ALL)
self.create_log = 1
@ -84,7 +86,7 @@ class ScheduledJobType(Document):
def log_status(self, status):
# log file
frappe.logger("scheduler").info('Scheduled Job {0}: {1} for {2}'.format(status, self.method, frappe.local.site))
frappe.logger("scheduler").info(f"Scheduled Job {status}: {self.method} for {frappe.local.site}")
self.update_scheduler_log(status)
def update_scheduler_log(self, status):
@ -111,28 +113,28 @@ class ScheduledJobType(Document):
@frappe.whitelist()
def execute_event(doc):
frappe.only_for('System Manager')
def execute_event(doc: str):
frappe.only_for("System Manager")
doc = json.loads(doc)
frappe.get_doc('Scheduled Job Type', doc.get('name')).enqueue(force=True)
frappe.get_doc("Scheduled Job Type", doc.get("name")).enqueue(force=True)
def run_scheduled_job(job_type):
'''This is a wrapper function that runs a hooks.scheduler_events method'''
def run_scheduled_job(job_type: str):
"""This is a wrapper function that runs a hooks.scheduler_events method"""
try:
frappe.get_doc('Scheduled Job Type', dict(method=job_type)).execute()
frappe.get_doc("Scheduled Job Type", dict(method=job_type)).execute()
except Exception:
print(frappe.get_traceback())
def sync_jobs(hooks=None):
def sync_jobs(hooks: Dict = None):
frappe.reload_doc("core", "doctype", "scheduled_job_type")
scheduler_events = hooks or frappe.get_hooks("scheduler_events")
all_events = insert_events(scheduler_events)
clear_events(all_events)
def insert_events(scheduler_events):
def insert_events(scheduler_events: Dict) -> List:
cron_jobs, event_jobs = [], []
for event_type in scheduler_events:
events = scheduler_events.get(event_type)
@ -144,7 +146,7 @@ def insert_events(scheduler_events):
return cron_jobs + event_jobs
def insert_cron_jobs(events):
def insert_cron_jobs(events: Dict) -> List:
cron_jobs = []
for cron_format in events:
for event in events.get(cron_format):
@ -153,25 +155,29 @@ def insert_cron_jobs(events):
return cron_jobs
def insert_event_jobs(events, event_type):
def insert_event_jobs(events: List, event_type: str) -> List:
event_jobs = []
for event in events:
event_jobs.append(event)
frequency = event_type.replace('_', ' ').title()
frequency = event_type.replace("_", " ").title()
insert_single_event(frequency, event)
return event_jobs
def insert_single_event(frequency, event, cron_format=None):
def insert_single_event(frequency: str, event: str, cron_format: str = None):
cron_expr = {"cron_format": cron_format} if cron_format else {}
doc = frappe.get_doc({
"doctype": "Scheduled Job Type",
"method": event,
"cron_format": cron_format,
"frequency": frequency
})
doc = frappe.get_doc(
{
"doctype": "Scheduled Job Type",
"method": event,
"cron_format": cron_format,
"frequency": frequency,
}
)
if not frappe.db.exists("Scheduled Job Type", {"method": event, "frequency": frequency, **cron_expr }):
if not frappe.db.exists(
"Scheduled Job Type", {"method": event, "frequency": frequency, **cron_expr}
):
try:
doc.insert()
except frappe.DuplicateEntryError:
@ -180,7 +186,9 @@ def insert_single_event(frequency, event, cron_format=None):
def clear_events(all_events: List):
for event in frappe.get_all("Scheduled Job Type", fields=["name", "method", "server_script"]):
for event in frappe.get_all(
"Scheduled Job Type", fields=["name", "method", "server_script"]
):
is_server_script = event.server_script
is_defined_in_hooks = event.method in all_events