Conversation
Contributor
Reviewer's GuideAdds a new CLI module to manage xcore API and Celery worker processes, introduces extensive unit tests for the xworker service and registry, and bumps the package version from 2.1.2 to 2.1.3. Sequence diagram for xcore worker CLI start commandsequenceDiagram
actor User
participant CLI as xcore_worker_CLI
participant Handler as handle_worker
participant StartCmd as _cmd_start
participant API as _start_api
participant Celery as _start_celery
participant Uvicorn as uvicorn_process
participant CeleryProc as celery_worker_process
User->>CLI: xcore worker start [--detach]
CLI->>Handler: handle_worker(args)
Handler->>StartCmd: _cmd_start(args)
StartCmd->>API: _start_api(args)
alt detach_mode
API->>Uvicorn: subprocess.Popen(..., start_new_session=True)
API->>API: _write_pid(PID_API, pid)
else foreground
API->>Uvicorn: subprocess.Popen(cmd)
end
StartCmd->>Celery: _start_celery(args)
alt detach_mode
Celery->>CeleryProc: subprocess.Popen(..., start_new_session=True)
Celery->>Celery: _write_pid(PID_CELERY, pid)
else foreground
Celery->>CeleryProc: subprocess.Popen(cmd)
end
opt foreground
StartCmd->>StartCmd: _is_running(pid) loop
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Contributor
|
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 7 security issues, 2 other issues, and left some high level feedback:
Security issues:
- Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
General comments:
- The
__resolve_celery_apphelper currently always returns the same hardcoded app path even when a config is loaded, so either remove the unused config/enabled check or actually support a configurable Celery app path from the configuration. - The CLI log commands rely on the external
tailexecutable (includingtail -f), which will fail on platforms without it (e.g. Windows); consider implementing the tail/follow logic in pure Python to avoid this dependency. - The beat process writes its own PID and log files (
beat.pid,beat.log) but the stop/status commands only handle API and Celery PIDs, so you may want to wire beat into the same lifecycle management (status/stop) or clearly separate its management to avoid orphaned processes.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `__resolve_celery_app` helper currently always returns the same hardcoded app path even when a config is loaded, so either remove the unused config/enabled check or actually support a configurable Celery app path from the configuration.
- The CLI log commands rely on the external `tail` executable (including `tail -f`), which will fail on platforms without it (e.g. Windows); consider implementing the tail/follow logic in pure Python to avoid this dependency.
- The beat process writes its own PID and log files (`beat.pid`, `beat.log`) but the stop/status commands only handle API and Celery PIDs, so you may want to wire beat into the same lifecycle management (status/stop) or clearly separate its management to avoid orphaned processes.
## Individual Comments
### Comment 1
<location path="xcore/cli/worker_cmd.py" line_range="102-105" />
<code_context>
+ return None
+
+
+def _resolve_celery_app(config_path: str | None) -> str:
+ """Retourne le chemin Celery app depuis la config ou le défaut."""
+ cfg = _load_config(config_path)
+ if cfg and cfg.services.xworker.enabled:
+ return "xcore.services.xworker.main:app"
+ return "xcore.services.xworker.main:app"
</code_context>
<issue_to_address>
**issue (bug_risk):** La condition sur la config n’influence jamais le chemin Celery retourné.
La fonction ignore donc complètement la configuration : `xworker.enabled` n’est jamais pris en compte et un éventuel autre chemin d’app reste inutilisé. Si l’objectif est de rendre le module Celery configurable, il faudrait dériver `celery_app` de `cfg.services.xworker` (par ex. `cfg.services.xworker.app_path`) ou retourner `None` / lever une erreur lorsque `xworker.enabled` est `False`. Sinon, la condition peut être supprimée pour éviter toute ambiguïté.
</issue_to_address>
### Comment 2
<location path="xcore/cli/worker_cmd.py" line_range="353-362" />
<code_context>
+ if target in ("all", "celery"):
+ targets.append(("Celery", LOG_CELERY))
+
+ if follow and len(targets) == 1:
+ label, log_path = targets[0]
+ if not log_path.exists():
+ console.print(
+ f"[yellow]⚠[/yellow] {log_path} introuvable — le service est-il démarré ?"
+ )
+ return
+ console.print(f"[dim]→ {log_path} (Ctrl+C pour quitter)[/dim]\n")
+ try:
+ proc = subprocess.Popen(["tail", "-f", "-n", str(lines), str(log_path)])
+ proc.wait()
+ except KeyboardInterrupt:
</code_context>
<issue_to_address>
**suggestion (bug_risk):** La dépendance à la commande `tail` limite la portabilité de la sous-commande `logs`.
Cette logique repose sur `tail -f`, indisponible ou différent sur certaines plateformes (notamment Windows sans environnement Unix). Il vaudrait mieux soit implémenter un `tail` minimal en Python (seek en fin de fichier, lecture en boucle), soit prévoir un fallback explicite lorsque `tail` est introuvable (`FileNotFoundError`), afin d’éviter des échecs silencieux ou des comportements inattendus hors Unix.
Suggested implementation:
```python
import subprocess
import time
```
```python
target = getattr(args, "target", "all")
lines = getattr(args, "lines", 50)
follow = getattr(args, "follow", False)
targets: list[tuple[str, Path]] = []
if target in ("all", "api"):
targets.append(("API", LOG_API))
if target in ("all", "celery"):
targets.append(("Celery", LOG_CELERY))
# Si on suit un seul fichier de log, on utilise tail -f lorsqu'il est disponible,
# avec un fallback Python lorsque la commande n'existe pas.
if follow and len(targets) == 1:
label, log_path = targets[0]
if not log_path.exists():
console.print(
f"[yellow]⚠[/yellow] {log_path} introuvable — le service est-il démarré ?"
)
return
console.print(f"[dim]→ {log_path} (Ctrl+C pour quitter)[/dim]\n")
def _python_tail_follow(path: Path, max_lines: int) -> None:
# Lecture des dernières lignes, puis suivi des nouvelles écritures.
with path.open("rb") as f:
# On lit les dernières `max_lines` lignes en remontant depuis la fin.
f.seek(0, 2)
file_size = f.tell()
block_size = 1024
data = b""
pos = file_size
while pos > 0 and data.count(b"\n") <= max_lines:
read_size = min(block_size, pos)
pos -= read_size
f.seek(pos, 0)
chunk = f.read(read_size)
data = chunk + data
if pos == 0:
break
lines_bytes = data.splitlines()[-max_lines:]
for line in lines_bytes:
console.print(line.decode(errors="replace"))
# Suivi continu
while True:
line = f.readline()
if not line:
time.sleep(0.5)
continue
console.print(line.decode(errors="replace"), end="")
try:
proc = subprocess.Popen(["tail", "-f", "-n", str(lines), str(log_path)])
proc.wait()
except FileNotFoundError:
# Fallback portable si la commande `tail` n'est pas disponible.
try:
_python_tail_follow(log_path, lines)
except KeyboardInterrupt:
pass
except KeyboardInterrupt:
pass
return
console.print("\n[yellow]⚠[/yellow] Arrêt en cours…")
finally:
for proc in procs:
if _is_running(proc.pid):
proc.terminate()
for proc in procs:
try:
proc.wait(timeout=8)
except subprocess.TimeoutExpired:
proc.kill()
```
1. Assurez-vous que ce bloc se trouve bien dans la fonction/commande qui gère la sous-commande `logs` (le `finally` suggère que d'autres parties de la fonction sont omises ici).
2. Si des conventions de logging ou d'affichage spécifiques existent (par exemple une fonction utilitaire pour afficher des logs), adaptez l'appel dans `_python_tail_follow` pour respecter ces conventions.
3. Si votre codebase a déjà un utilitaire générique pour suivre un fichier (file watcher, etc.), remplacez `_python_tail_follow` par un appel à cet utilitaire pour éviter la duplication.
</issue_to_address>
### Comment 3
<location path="xcore/cli/worker_cmd.py" line_range="146-151" />
<code_context>
proc = subprocess.Popen(
cmd,
stdout=log_file,
stderr=log_file,
start_new_session=True,
)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 4
<location path="xcore/cli/worker_cmd.py" line_range="160" />
<code_context>
return subprocess.Popen(cmd)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 5
<location path="xcore/cli/worker_cmd.py" line_range="209-214" />
<code_context>
proc = subprocess.Popen(
cmd,
stdout=log_file,
stderr=log_file,
start_new_session=True,
)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 6
<location path="xcore/cli/worker_cmd.py" line_range="223" />
<code_context>
return subprocess.Popen(cmd)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 7
<location path="xcore/cli/worker_cmd.py" line_range="463" />
<code_context>
result = subprocess.run(cmd, capture_output=True, text=True)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 8
<location path="xcore/cli/worker_cmd.py" line_range="498-503" />
<code_context>
proc = subprocess.Popen(
cmd,
stdout=log_file,
stderr=log_file,
start_new_session=True,
)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 9
<location path="xcore/cli/worker_cmd.py" line_range="513" />
<code_context>
proc = subprocess.Popen(cmd)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+102
to
+105
| def _resolve_celery_app(config_path: str | None) -> str: | ||
| """Retourne le chemin Celery app depuis la config ou le défaut.""" | ||
| cfg = _load_config(config_path) | ||
| if cfg and cfg.services.xworker.enabled: |
Contributor
There was a problem hiding this comment.
issue (bug_risk): La condition sur la config n’influence jamais le chemin Celery retourné.
La fonction ignore donc complètement la configuration : xworker.enabled n’est jamais pris en compte et un éventuel autre chemin d’app reste inutilisé. Si l’objectif est de rendre le module Celery configurable, il faudrait dériver celery_app de cfg.services.xworker (par ex. cfg.services.xworker.app_path) ou retourner None / lever une erreur lorsque xworker.enabled est False. Sinon, la condition peut être supprimée pour éviter toute ambiguïté.
Comment on lines
+353
to
+362
| if follow and len(targets) == 1: | ||
| label, log_path = targets[0] | ||
| if not log_path.exists(): | ||
| console.print( | ||
| f"[yellow]⚠[/yellow] {log_path} introuvable — le service est-il démarré ?" | ||
| ) | ||
| return | ||
| console.print(f"[dim]→ {log_path} (Ctrl+C pour quitter)[/dim]\n") | ||
| try: | ||
| proc = subprocess.Popen(["tail", "-f", "-n", str(lines), str(log_path)]) |
Contributor
There was a problem hiding this comment.
suggestion (bug_risk): La dépendance à la commande tail limite la portabilité de la sous-commande logs.
Cette logique repose sur tail -f, indisponible ou différent sur certaines plateformes (notamment Windows sans environnement Unix). Il vaudrait mieux soit implémenter un tail minimal en Python (seek en fin de fichier, lecture en boucle), soit prévoir un fallback explicite lorsque tail est introuvable (FileNotFoundError), afin d’éviter des échecs silencieux ou des comportements inattendus hors Unix.
Suggested implementation:
import subprocess
import time target = getattr(args, "target", "all")
lines = getattr(args, "lines", 50)
follow = getattr(args, "follow", False)
targets: list[tuple[str, Path]] = []
if target in ("all", "api"):
targets.append(("API", LOG_API))
if target in ("all", "celery"):
targets.append(("Celery", LOG_CELERY))
# Si on suit un seul fichier de log, on utilise tail -f lorsqu'il est disponible,
# avec un fallback Python lorsque la commande n'existe pas.
if follow and len(targets) == 1:
label, log_path = targets[0]
if not log_path.exists():
console.print(
f"[yellow]⚠[/yellow] {log_path} introuvable — le service est-il démarré ?"
)
return
console.print(f"[dim]→ {log_path} (Ctrl+C pour quitter)[/dim]\n")
def _python_tail_follow(path: Path, max_lines: int) -> None:
# Lecture des dernières lignes, puis suivi des nouvelles écritures.
with path.open("rb") as f:
# On lit les dernières `max_lines` lignes en remontant depuis la fin.
f.seek(0, 2)
file_size = f.tell()
block_size = 1024
data = b""
pos = file_size
while pos > 0 and data.count(b"\n") <= max_lines:
read_size = min(block_size, pos)
pos -= read_size
f.seek(pos, 0)
chunk = f.read(read_size)
data = chunk + data
if pos == 0:
break
lines_bytes = data.splitlines()[-max_lines:]
for line in lines_bytes:
console.print(line.decode(errors="replace"))
# Suivi continu
while True:
line = f.readline()
if not line:
time.sleep(0.5)
continue
console.print(line.decode(errors="replace"), end="")
try:
proc = subprocess.Popen(["tail", "-f", "-n", str(lines), str(log_path)])
proc.wait()
except FileNotFoundError:
# Fallback portable si la commande `tail` n'est pas disponible.
try:
_python_tail_follow(log_path, lines)
except KeyboardInterrupt:
pass
except KeyboardInterrupt:
pass
return
console.print("\n[yellow]⚠[/yellow] Arrêt en cours…")
finally:
for proc in procs:
if _is_running(proc.pid):
proc.terminate()
for proc in procs:
try:
proc.wait(timeout=8)
except subprocess.TimeoutExpired:
proc.kill()- Assurez-vous que ce bloc se trouve bien dans la fonction/commande qui gère la sous-commande
logs(lefinallysuggère que d'autres parties de la fonction sont omises ici). - Si des conventions de logging ou d'affichage spécifiques existent (par exemple une fonction utilitaire pour afficher des logs), adaptez l'appel dans
_python_tail_followpour respecter ces conventions. - Si votre codebase a déjà un utilitaire générique pour suivre un fichier (file watcher, etc.), remplacez
_python_tail_followpar un appel à cet utilitaire pour éviter la duplication.
Comment on lines
+146
to
+151
| proc = subprocess.Popen( | ||
| cmd, | ||
| stdout=log_file, | ||
| stderr=log_file, | ||
| start_new_session=True, | ||
| ) |
Contributor
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
| return proc | ||
| else: | ||
| console.print(f" [cyan]→[/cyan] API [dim]{' '.join(cmd[2:])}[/dim]") | ||
| return subprocess.Popen(cmd) |
Contributor
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
Comment on lines
+209
to
+214
| proc = subprocess.Popen( | ||
| cmd, | ||
| stdout=log_file, | ||
| stderr=log_file, | ||
| start_new_session=True, | ||
| ) |
Contributor
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
| return proc | ||
| else: | ||
| console.print(f" [cyan]→[/cyan] Celery [dim]{' '.join(cmd[2:])}[/dim]") | ||
| return subprocess.Popen(cmd) |
Contributor
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
| ] | ||
|
|
||
| console.print(f"[yellow]⚠[/yellow] Purge de la file [bold]{queue}[/bold]…") | ||
| result = subprocess.run(cmd, capture_output=True, text=True) |
Contributor
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
Comment on lines
+498
to
+503
| proc = subprocess.Popen( | ||
| cmd, | ||
| stdout=log_file, | ||
| stderr=log_file, | ||
| start_new_session=True, | ||
| ) |
Contributor
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
| console.print(f" [cyan]→[/cyan] Beat [dim]{' '.join(cmd[2:])}[/dim]") | ||
| console.print("[dim]Ctrl+C pour arrêter[/dim]\n") | ||
| try: | ||
| proc = subprocess.Popen(cmd) |
Contributor
There was a problem hiding this comment.
security (python.lang.security.audit.dangerous-subprocess-use-audit): Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
Source: opengrep
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Bump xcore to version 2.1.3 and introduce a CLI worker management command with comprehensive tests for the Celery-based worker service and its configuration.
New Features:
Enhancements:
Tests: