mirror of
https://github.com/suitenumerique/meet.git
synced 2026-08-13 12:17:24 +00:00
davd-gzl
a67467b193
The SonarCloud gate fails on main, so every commit lands red, and gh run list hides it: it lists only Actions workflows, and the failure is an app check run. Reliability rests on one bug, in test_file_service.py, which wrapped an assertion in an except Exception re-raised through pytest.fail. Removing it takes the rating from D to A. Two pieces of debt ride along. The SDK callback id now comes from crypto.getRandomValues, since it guards an endpoint with no auth. And core/tasks gets the __init__.py that lets pylint see it, with the debt that exposes, which is why #1533 fails lint-back.
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
"""
|
|
Celery task decorator that degrades to a synchronous call when Celery is off.
|
|
"""
|
|
|
|
# The Celery app is imported lazily so that importing this module does not pull
|
|
# in Celery when CELERY_ENABLED is false.
|
|
# ruff: noqa: PLC0415
|
|
# pylint: disable=import-outside-toplevel
|
|
|
|
from django.conf import settings
|
|
|
|
|
|
def task(*d_args, **d_kwargs):
|
|
"""
|
|
Decorator compatible with Celery's @app.task, but works without Celery.
|
|
|
|
If Celery is available, returns a real Celery task.
|
|
If not, returns the original function and provides `.delay()`/`.apply_async()`
|
|
as synchronous fallbacks (so existing call sites don't break).
|
|
|
|
Notes:
|
|
Mostly LLM-generated.
|
|
"""
|
|
|
|
def _fallback_wrap(func):
|
|
def delay(*args, **kwargs):
|
|
return func(*args, **kwargs)
|
|
|
|
def apply_async(args=None, kwargs=None, **_options):
|
|
return func(*(args or ()), **(kwargs or {}))
|
|
|
|
func.delay = delay
|
|
func.apply_async = apply_async
|
|
return func
|
|
|
|
# Handle bare decorator usage: @task
|
|
if len(d_args) == 1 and callable(d_args[0]) and not d_kwargs:
|
|
func = d_args[0]
|
|
if settings.CELERY_ENABLED:
|
|
from meet.celery_app import app as _celery_app
|
|
|
|
return _celery_app.task(func)
|
|
return _fallback_wrap(func)
|
|
|
|
# Handle parameterized usage: @task(...), e.g. @task(bind=True)
|
|
def _decorate(func):
|
|
if settings.CELERY_ENABLED:
|
|
from meet.celery_app import app as _celery_app
|
|
|
|
return _celery_app.task(*d_args, **d_kwargs)(func)
|
|
return _fallback_wrap(func)
|
|
|
|
return _decorate
|
|
|
|
|
|
__all__ = ("task",)
|