♻️(core) move app ready code to functions

For readability, we move the code block from the `ready` method to a
dedicated function.

This will allow to add more things to do in the `ready` with more focus.
This commit is contained in:
Quentin BEY
2025-03-26 09:51:30 +01:00
committed by Sabrina Demagny
parent 2502ff0c99
commit 4ced342062

View File

@@ -14,6 +14,49 @@ from people.celery_app import app as celery_app
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _register_management_commands_as_task():
"""
Register management command which are enabled via MANAGEMENT_COMMAND_AS_TASK setting.
"""
for command_name in settings.MANAGEMENT_COMMAND_AS_TASK:
# Check if the command is a valid management command
try:
app_name = get_commands()[command_name]
except KeyError:
logging.error("Command %s is not a valid management command.", command_name)
continue
command_full_name = ".".join([app_name, command_name])
# Create a closure to capture the current value of command_full_name and command_name
def create_task(cmd_name, cmd_full_name):
@celery_app.task(name=cmd_full_name)
def task_wrapper(*command_args, **command_options):
stdout = TeeStringIO(logging.getLogger(cmd_full_name).info)
stderr = TeeStringIO(logging.getLogger(cmd_full_name).error)
call_command(
cmd_name,
*command_args,
no_color=True,
stdout=stdout,
stderr=stderr,
**command_options,
)
stdout.seek(0)
stderr.seek(0)
return {
"stdout": str(stdout.read()),
"stderr": str(stderr.read()),
}
return task_wrapper
# Create the task with the current values
create_task(command_name, command_full_name)
class CoreConfig(AppConfig): class CoreConfig(AppConfig):
"""Configuration class for the People core app.""" """Configuration class for the People core app."""
@@ -22,45 +65,5 @@ class CoreConfig(AppConfig):
verbose_name = _("People core application") verbose_name = _("People core application")
def ready(self): def ready(self):
""" """Run when the application is ready."""
Register management command which are enabled via MANAGEMENT_COMMAND_AS_TASK setting. _register_management_commands_as_task()
"""
for command_name in settings.MANAGEMENT_COMMAND_AS_TASK:
# Check if the command is a valid management command
try:
app_name = get_commands()[command_name]
except KeyError:
logging.error(
"Command %s is not a valid management command.", command_name
)
continue
command_full_name = ".".join([app_name, command_name])
# Create a closure to capture the current value of command_full_name and command_name
def create_task(cmd_name, cmd_full_name):
@celery_app.task(name=cmd_full_name)
def task_wrapper(*command_args, **command_options):
stdout = TeeStringIO(logging.getLogger(cmd_full_name).info)
stderr = TeeStringIO(logging.getLogger(cmd_full_name).error)
call_command(
cmd_name,
*command_args,
no_color=True,
stdout=stdout,
stderr=stderr,
**command_options,
)
stdout.seek(0)
stderr.seek(0)
return {
"stdout": str(stdout.read()),
"stderr": str(stderr.read()),
}
return task_wrapper
# Create the task with the current values
create_task(command_name, command_full_name)