Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
156 lines
5.3 KiB
Python
156 lines
5.3 KiB
Python
"""ARQ job queue integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from arq import create_pool
|
|
from arq.connections import ArqRedis, RedisSettings
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Global ARQ pool singleton ────────────────────────────────────────────────
|
|
_job_pool: ArqRedis | None = None
|
|
|
|
|
|
async def init_job_pool() -> ArqRedis:
|
|
"""Create and store the global ARQ job pool.
|
|
|
|
Called once during app lifespan startup so every subsequent enqueue
|
|
reuses the same connection instead of opening a new one per call.
|
|
"""
|
|
global _job_pool
|
|
if _job_pool is not None:
|
|
logger.warning("init_job_pool() called but pool already initialized")
|
|
return _job_pool
|
|
settings = get_settings()
|
|
redis_settings = RedisSettings.from_dsn(settings.redis_url)
|
|
_job_pool = await create_pool(redis_settings)
|
|
logger.info("Global ARQ job pool initialized")
|
|
return _job_pool
|
|
|
|
|
|
async def close_job_pool() -> None:
|
|
"""Close the global ARQ job pool. Called during app lifespan shutdown."""
|
|
global _job_pool
|
|
if _job_pool is not None:
|
|
await _job_pool.close()
|
|
_job_pool = None
|
|
logger.info("Global ARQ job pool closed")
|
|
|
|
|
|
async def get_job_pool() -> ArqRedis:
|
|
"""Return the global ARQ job pool singleton.
|
|
|
|
If init_job_pool() has not been called yet (e.g. during testing),
|
|
a new pool is created lazily so callers always get a working connection.
|
|
"""
|
|
global _job_pool
|
|
if _job_pool is None:
|
|
settings = get_settings()
|
|
redis_settings = RedisSettings.from_dsn(settings.redis_url)
|
|
_job_pool = await create_pool(redis_settings)
|
|
logger.debug("ARQ job pool created lazily (init_job_pool not called)")
|
|
return _job_pool
|
|
|
|
|
|
async def enqueue_job(job_name: str, *args: Any, **kwargs: Any) -> str | None:
|
|
"""Enqueue a background job. Returns job_id or None."""
|
|
pool = await get_job_pool()
|
|
job = await pool.enqueue_job(job_name, *args, **kwargs)
|
|
if job:
|
|
return job.job_id
|
|
return None
|
|
|
|
|
|
async def get_job_status(job_id: str) -> dict[str, Any] | None:
|
|
"""Get the status of a background job."""
|
|
pool = await get_job_pool()
|
|
job_info = await pool.job_info(job_id)
|
|
if job_info is None:
|
|
return None
|
|
return {
|
|
"job_id": job_id,
|
|
"status": str(job_info.status),
|
|
"result": job_info.result,
|
|
"enqueue_time": job_info.enqueue_time.isoformat() if job_info.enqueue_time else None,
|
|
"start_time": job_info.start_time.isoformat() if job_info.start_time else None,
|
|
"finish_time": job_info.finish_time.isoformat() if job_info.finish_time else None,
|
|
}
|
|
|
|
|
|
# ── Password Reset Email Job ─────────────────────────────────────────────────
|
|
|
|
async def send_password_reset_email(
|
|
ctx: dict[str, Any],
|
|
*,
|
|
user_id: str,
|
|
email: str,
|
|
raw_token: str,
|
|
expires_at: str,
|
|
) -> None:
|
|
"""Send a password reset email via SMTP.
|
|
|
|
This is an ARQ worker function. It is registered with the job registry
|
|
so the worker can execute it when the auth service enqueues it.
|
|
"""
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
|
|
import aiosmtplib
|
|
|
|
settings = get_settings()
|
|
|
|
# Build the reset URL
|
|
reset_url = f"{settings.frontend_url.rstrip('/')}/reset-password?token={raw_token}"
|
|
|
|
# Build the email
|
|
msg = MIMEMultipart("alternative")
|
|
msg["From"] = settings.smtp_from_email
|
|
msg["To"] = email
|
|
msg["Subject"] = "LeoCRM — Passwort zurücksetzen"
|
|
|
|
text_body = (
|
|
f"Sie haben angefordert, Ihr Passwort zurückzusetzen.\n\n"
|
|
f"Klicken Sie auf den folgenden Link, um ein neues Passwort zu setzen:\n"
|
|
f"{reset_url}\n\n"
|
|
f"Dieser Link ist gültig bis {expires_at}.\n\n"
|
|
f"Falls Sie diese Anfrage nicht gestellt haben, können Sie diese\n"
|
|
f"E-Mail ignorieren. Ihr Passwort bleibt unverändert.\n"
|
|
)
|
|
html_body = (
|
|
f"<html><body>"
|
|
f"<h2>Passwort zurücksetzen</h2>"
|
|
f"<p>Sie haben angefordert, Ihr Passwort zurückzusetzen.</p>"
|
|
f"<p><a href=\"{reset_url}\">Passwort jetzt zurücksetzen</a></p>"
|
|
f"<p>Dieser Link ist gültig bis {expires_at}.</p>"
|
|
f"<p>Falls Sie diese Anfrage nicht gestellt haben, können Sie diese "
|
|
f"E-Mail ignorieren. Ihr Passwort bleibt unverändert.</p>"
|
|
f"</body></html>"
|
|
)
|
|
msg.attach(MIMEText(text_body, "plain", "utf-8"))
|
|
msg.attach(MIMEText(html_body, "html", "utf-8"))
|
|
|
|
# Send via SMTP (port 465 = implicit TLS, port 587 = STARTTLS)
|
|
use_tls = settings.smtp_port == 465
|
|
start_tls = settings.smtp_use_tls and not use_tls
|
|
await aiosmtplib.send(
|
|
msg,
|
|
hostname=settings.smtp_host,
|
|
port=settings.smtp_port,
|
|
username=settings.smtp_username,
|
|
password=settings.smtp_password,
|
|
use_tls=use_tls,
|
|
start_tls=start_tls,
|
|
)
|
|
logger.info("Password reset email sent to %s for user %s", email, user_id)
|
|
|
|
|
|
# Register the job so the worker can find it
|
|
from app.core.job_registry import register_job # noqa: E402
|
|
|
|
register_job("send_password_reset_email", send_password_reset_email)
|