Return job class from RQ settings, otherwise return Job. If `job_class` is not None, it is used as an override (can be python import path as string).
(job_class: Optional[Union[str, type[Job]]] = None)
| 6 | |
| 7 | |
| 8 | def get_job_class(job_class: Optional[Union[str, type[Job]]] = None) -> type[Job]: |
| 9 | """ |
| 10 | Return job class from RQ settings, otherwise return Job. |
| 11 | If `job_class` is not None, it is used as an override (can be |
| 12 | python import path as string). |
| 13 | """ |
| 14 | RQ = getattr(settings, 'RQ', {}) |
| 15 | |
| 16 | if not job_class: |
| 17 | job_class = RQ.get('JOB_CLASS') |
| 18 | |
| 19 | # Ensure we never have None (in case someone explicitly sets JOB_CLASS to None) |
| 20 | if not job_class: |
| 21 | job_class = Job |
| 22 | |
| 23 | if isinstance(job_class, str): |
| 24 | job_class = cast(type[Job], import_attribute(job_class)) |
| 25 | return job_class |
no outgoing calls
searching dependent graphs…