From 531b9cb10e4388e1d8a6cc80070b93bfea0a36ce Mon Sep 17 00:00:00 2001 From: Frank Agerholm Date: Fri, 16 Jan 2026 20:51:40 +0100 Subject: [PATCH 1/2] =?UTF-8?q?Basisimplementierung=20f=C3=BCr=20Mailversa?= =?UTF-8?q?ndt.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api.py | 65 ++++++----- app/config.py | 33 +++++- app/db.py | 6 +- app/job_registry.py | 5 +- app/models.py | 23 +++- app/routers/mail.py | 79 +++++++++++++ app/schema_utils.py | 26 +++++ app/schemas.py | 17 ++- app/workers/mail_send_worker.py | 67 +++++++++++ app/workers/mail_store_worker.py | 70 +++++++++++ .../versions/20260116_create_messages.py | 28 +++++ ...7c598_merge_message_branch_and_baseline.py | 28 +++++ poetry.lock | 109 +++++++++++++++++- pyproject.toml | 5 +- 14 files changed, 520 insertions(+), 41 deletions(-) create mode 100644 app/routers/mail.py create mode 100644 app/schema_utils.py create mode 100644 app/workers/mail_send_worker.py create mode 100644 app/workers/mail_store_worker.py create mode 100644 migrations/versions/20260116_create_messages.py create mode 100644 migrations/versions/918e1dd7c598_merge_message_branch_and_baseline.py diff --git a/app/api.py b/app/api.py index a94c581..5d47b2e 100644 --- a/app/api.py +++ b/app/api.py @@ -1,24 +1,46 @@ from fastapi import FastAPI, Depends, HTTPException +from contextlib import asynccontextmanager from sqlalchemy.ext.asyncio import AsyncSession -from app.db import Base, engine, SessionLocal +from app.db import Base, engine, get_session from app.schemas import CreateJobRequest, JobInfo, TaskInfo +from app.schema_utils import job_to_schema from app.services import create_job, get_job_with_tasks from app.models import Job, Task + +from app.routers.mail import router as mail_router + from typing import List from uuid import UUID import re, idna -app = FastAPI(title="Task Queue API", version="0.3.0") +@asynccontextmanager +async def lifespan(app: FastAPI): + # STARTUP Beispiele + # await init_db_pool + # await init_broker() -async def get_session() -> AsyncSession: - async with SessionLocal() as session: - yield session + # Optionale Werte Beispiele + # app.state.db_pool = get_db_pool + # app.state.broker = get_broker() -@app.on_event("startup") -async def on_startup(): - async with engine.begin() as conn: + # Beispiel: DB-Update beim Startup + #await conn.run_sync(Base.metadata.create_all) + + try: + yield + finally: + # SHUTDOWN Beispiele + # await close_broker() + # await close_db_pool() pass - #await conn.run_sync(Base.metadata.create_all) + +app = FastAPI( + title="Task Queue API", + version="0.3.0", + lifespan=lifespan) + +app.include_router(mail_router) + DOMAIN_RE = re.compile(r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(? JobInfo: - tasks: List[TaskInfo] = [ - TaskInfo( - id=t.id, - name=t.name, - status=t.status.value if hasattr(t.status, "value") else str(t.status), - retries=t.retries, - max_retries=t.max_retries, - result=t.result, - error=t.error, - ) - for t in job.tasks - ] - return JobInfo( - id=job.id, - job_type=job.job_type, - status=job.status.value if hasattr(job.status, "value") else str(job.status), - domain=job.domain, - payload=job.payload, - tasks=tasks, - ) diff --git a/app/config.py b/app/config.py index bfa4dc1..3cf63c0 100644 --- a/app/config.py +++ b/app/config.py @@ -1,6 +1,28 @@ -from pydantic import BaseModel +from pydantic import BaseModel, Field, AnyUrl +from typing import Optional import os +def str_to_bool(value): + """ + Convert a string to a boolean. + Accepts common truthy/falsey string values. + Raises ValueError for unrecognized inputs. + """ + if not isinstance(value, str): + raise TypeError("Input must be a string.") + + value_lower = value.strip().lower() + + truthy = {"true", "1", "yes", "y", "on"} + falsey = {"false", "0", "no", "n", "off"} + + if value_lower in truthy: + return True + elif value_lower in falsey: + return False + else: + raise ValueError(f"Cannot convert '{value}' to boolean.") + class Settings(BaseModel): DATABASE_URL: str = os.getenv("DATABASE_URL", "postgresql+asyncpg://postgres:postgres@localhost:5432/tasks") RABBITMQ_URL: str = os.getenv("RABBITMQ_URL", "amqp://guest:guest@localhost/") @@ -12,4 +34,11 @@ class Settings(BaseModel): SMTP_TIMEOUT: float = float(os.getenv("SMTP_TIMEOUT", "10.0")) HTTP_TIMEOUT: float = float(os.getenv("HTTP_TIMEOUT", "8.0")) -settings = Settings() \ No newline at end of file + smtp_host: str = os.getenv("SMTP_HOST", "localhost") + smtp_port: int = os.getenv("SMTP_PORT", "587") + smtp_user: Optional[str] = os.getenv("SMTP_USER") + smtp_password: Optional[str] = os.getenv("SMTP_PASSWORD") + smtp_starttls: bool = str_to_bool(os.getenv("SMTP_STARTTLS", "true")) + smtp_sender_fallback: str = os.getenv("SMTP_SENDER", "noreply@example.com") + +settings = Settings() diff --git a/app/db.py b/app/db.py index 90b5403..7cb53f6 100644 --- a/app/db.py +++ b/app/db.py @@ -5,5 +5,9 @@ from app.config import settings engine = create_async_engine(settings.DATABASE_URL, echo=False, future=True) SessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession) +async def get_session() -> AsyncSession: + async with SessionLocal() as session: + yield session + class Base(DeclarativeBase): - pass \ No newline at end of file + pass diff --git a/app/job_registry.py b/app/job_registry.py index 3ed6515..4472a83 100644 --- a/app/job_registry.py +++ b/app/job_registry.py @@ -9,6 +9,9 @@ JOB_GRAPH: Dict[str, Dict[str, List[str]]] = { "CertValidity": ["MXValidation"], "MTASTSCheck": [], "TLSRPTCheck": [], + }, + "MailSend": { + "SendClean": [], } } @@ -19,4 +22,4 @@ def list_tasks_for_job(job_type: str) -> List[str]: return list(JOB_GRAPH.get(job_type, {}).keys()) def list_dependencies(job_type: str, task_name: str) -> List[str]: - return JOB_GRAPH.get(job_type, {}).get(task_name, []) \ No newline at end of file + return JOB_GRAPH.get(job_type, {}).get(task_name, []) diff --git a/app/models.py b/app/models.py index f137a2c..e9e3f23 100644 --- a/app/models.py +++ b/app/models.py @@ -1,7 +1,9 @@ import enum, uuid +from datetime import datetime from sqlalchemy import String, Enum, ForeignKey, Integer, DateTime, JSON +from sqlalchemy import LargeBinary, Index from sqlalchemy.orm import relationship, Mapped, mapped_column -from sqlalchemy.dialects.postgresql import UUID +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP from sqlalchemy.sql import func from app.db import Base @@ -52,3 +54,22 @@ class TaskDependency(Base): __tablename__ = "task_dependencies" task_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="CASCADE"), primary_key=True) depends_on_task_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="CASCADE"), primary_key=True) + + +class Message(Base): + __tablename__ = "messages" + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + mailbox_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False, index=True) + message_id: Mapped[str | None]= mapped_column(String(512), index=True, nullable=True) + subject: Mapped[str | None] = mapped_column(String(2048)) + from_addr: Mapped[str | None] = mapped_column(String(2048)) + to_addr: Mapped[str | None] = mapped_column(String(4096)) + date_header: Mapped[datetime | None] = mapped_column(TIMESTAMP(timezone=True), nullable=True) + received_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), default=datetime.utcnow, nullable=False) + raw_eml: Mapped[bytes] = mapped_column(LargeBinary, nullable=False) + +Index("ix_messages_date", Message.date_header) +Index("ix_messages_message_id", Message.message_id) + + diff --git a/app/routers/mail.py b/app/routers/mail.py new file mode 100644 index 0000000..3842ab7 --- /dev/null +++ b/app/routers/mail.py @@ -0,0 +1,79 @@ +# app/routers/mail.py +from __future__ import annotations +import base64 +from uuid import UUID +from fastapi import APIRouter, UploadFile, File, Form, Depends, status, HTTPException +from fastapi.encoders import jsonable_encoder +from sqlalchemy.ext.asyncio import AsyncSession +from app.db import get_session +from app.schemas import SendTestMailRequest, UploadMailResponse +from app.schema_utils import job_to_schema +from app.task_enqueue import enqueue_job_with_single_task +from app.services import create_job, get_job_with_tasks +from app.job_registry import routing_key +from email import policy +from email.parser import BytesParser + +router = APIRouter(prefix="/api/v1/mail", tags=["mail"]) + +# TODO: vorhandene Auth-Dependency einhängen +async def api_key_auth(): + return True + +def _guess_domain_from_addresses(addresses: str | None) -> str: + if not addresses: + return "unknown.local" + # nimmt erste Adresse, extrahiert Domain + first = addresses.split(",")[0].strip() + if "@" in first: + return first.split("@")[-1].strip().lower() + return "unknown.local" + +@router.post("/send", status_code=status.HTTP_202_ACCEPTED) +async def send_test_mail( + req: SendTestMailRequest, + _=Depends(api_key_auth), + session: AsyncSession = Depends(get_session), +): + if not req.recipients: + raise HTTPException(status_code=400, detail="At least one recipient required.") + domain = req.recipients[0].split("@")[-1].lower() + job = await create_job( + session, + job_type="MailSend", + domain=domain, + payload=req.model_dump(mode="json") + ) + job = await get_job_with_tasks(session, job.id) + return job_to_schema(job) + +@router.post("/upload", response_model=UploadMailResponse, status_code=status.HTTP_202_ACCEPTED) +async def upload_mail( + mailbox_id: UUID = Form(...), + eml: UploadFile = File(...), + _=Depends(api_key_auth), + session: AsyncSession = Depends(get_session), +): + raw = await eml.read() + # Domain für Job.domain aus To/From ableiten + try: + msg = BytesParser(policy=policy.default).parsebytes(raw) + domain = _guess_domain_from_addresses(msg.get("To")) or _guess_domain_from_addresses(msg.get("From")) + except Exception: + domain = "unknown.local" + + payload = { + "mailbox_id": str(mailbox_id), + "raw_eml_b64": base64.b64encode(raw).decode("ascii"), + } + job_id, task_id = await enqueue_job_with_single_task( + session, + job_type="mail.store", + domain=domain, + payload=payload, + task_name="mail.store", + routing_key="mail.store", + ) + return UploadMailResponse(id=UUID(task_id), message_id=None) + + diff --git a/app/schema_utils.py b/app/schema_utils.py new file mode 100644 index 0000000..8bd4897 --- /dev/null +++ b/app/schema_utils.py @@ -0,0 +1,26 @@ + +from typing import List +from app.schemas import JobInfo, TaskInfo +from app.models import Job, Task + +def job_to_schema(job: Job) -> JobInfo: + tasks: List[TaskInfo] = [ + TaskInfo( + id=t.id, + name=t.name, + status=t.status.value if hasattr(t.status, "value") else str(t.status), + retries=t.retries, + max_retries=t.max_retries, + result=t.result, + error=t.error, + ) + for t in job.tasks + ] + return JobInfo( + id=job.id, + job_type=job.job_type, + status=job.status.value if hasattr(job.status, "value") else str(job.status), + domain=job.domain, + payload=job.payload, + tasks=tasks, + ) diff --git a/app/schemas.py b/app/schemas.py index 6f0c9cf..3ce5ff3 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -1,4 +1,4 @@ -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, EmailStr from typing import Optional, List from uuid import UUID @@ -22,4 +22,17 @@ class JobInfo(BaseModel): status: str domain: str payload: dict | None - tasks: List[TaskInfo] \ No newline at end of file + tasks: List[TaskInfo] + +class SendTestMailRequest(BaseModel): + mailbox_id: UUID + recipients: List[EmailStr] + subject: str = Field(max_length=998) + body_text: Optional[str] = None + body_html: Optional[str] = None + # Anhänge könnt ihr später ergänzen + +class UploadMailResponse(BaseModel): + id: UUID + message_id: Optional[str] = None + diff --git a/app/workers/mail_send_worker.py b/app/workers/mail_send_worker.py new file mode 100644 index 0000000..40d6d9d --- /dev/null +++ b/app/workers/mail_send_worker.py @@ -0,0 +1,67 @@ + +# app/workers/mail_send_worker.py +from __future__ import annotations +import smtplib, ssl, email.utils +from email.message import EmailMessage +from typing import List, Optional +from pydantic import BaseModel, EmailStr +from app.workers.base_worker import BaseWorker +from app.config import Settings +from app.db import SessionLocal # <- AsyncSession factory +from app.models import Task, TaskStatus +from sqlalchemy import update +from app.job_registry import routing_key +from app.config import settings + +class _SendPayload(BaseModel): + mailbox_id: str + recipients: List[EmailStr] + subject: str + body_text: Optional[str] = None + body_html: Optional[str] = None + sender: Optional[EmailStr] = None + +class MailSendWorker(BaseWorker): + + async def process(self, payload: dict, **meta) -> dict: + """ + Erwartet: task.job.payload (JSON) mit Feldern wie in _SendPayload. + Setzt Task-Status/Result analog zu eurem BaseWorker-Pattern. + """ + p = _SendPayload(**payload) + + sender = str(p.sender or settings.smtp_sender_fallback) + msg = EmailMessage() + msg["From"] = sender + msg["To"] = ", ".join([str(r) for r in p.recipients]) + msg["Subject"] = p.subject + msg["Date"] = email.utils.formatdate(localtime=True) + msg["Message-ID"] = email.utils.make_msgid() + + if p.body_html and p.body_text: + msg.set_content(p.body_text) + msg.add_alternative(p.body_html, subtype="html") + elif p.body_html: + msg.add_alternative(p.body_html, subtype="html") + else: + msg.set_content(p.body_text or "") + + # SMTP – synchrones I/O im Worker-Kontext ist i.d.R. ok. + context = ssl.create_default_context() + with smtplib.SMTP(settings.smtp_host, settings.smtp_port, timeout=30) as server: + if settings.smtp_starttls: + server.starttls(context=context) + if settings.smtp_user and settings.smtp_password: + server.login(settings.smtp_user, settings.smtp_password) + server.send_message(msg) + + return {"message_id": msg["Message-ID"]} + +async def main(): + worker =MailSendWorker(routing_key("MailSend", "SendClean"), worker_id="mail-send-clean-1") + await worker.run() + +if __name__ == "__main__": + import asyncio as _a + _a.run(main()) + diff --git a/app/workers/mail_store_worker.py b/app/workers/mail_store_worker.py new file mode 100644 index 0000000..3de8379 --- /dev/null +++ b/app/workers/mail_store_worker.py @@ -0,0 +1,70 @@ + +# app/workers/mail_store_worker.py +from __future__ import annotations +import base64, email.utils +from email import policy +from email.parser import BytesParser +from datetime import datetime +from uuid import UUID +from sqlalchemy import select, update +from app.workers.base_worker import BaseWorker +from app.db import SessionLocal +from app.models import Message, Task, TaskStatus + +class MailStoreWorker(BaseWorker): + routing_key = "mail.store" + + async def handle(self, task) -> dict: + payload = task.job.payload # payload liegt auf dem Job + raw = base64.b64decode(payload["raw_eml_b64"]) + mailbox_id = UUID(payload["mailbox_id"]) + + msg = BytesParser(policy=policy.default).parsebytes(raw) + message_id = msg.get("Message-ID") + subject = msg.get("Subject") + from_addr = msg.get("From") + to_addr = msg.get("To") + dt = None + try: + dt = email.utils.parsedate_to_datetime(msg.get("Date")) if msg.get("Date") else None + except Exception: + pass + + async with SessionLocal() as session: + # Idempotenz: bei Message-ID prüfen + if message_id: + q = select(Message).where(Message.message_id == message_id) + existing = (await session.execute(q)).scalar_one_or_none() + if existing: + result = {"id": str(existing.id), "message_id": message_id, "dedup": True} + await session.execute( + update(Task) + .where(Task.id == task.id) + .values(status=TaskStatus.FINISHED, result=result, worker_id=self.worker_id) + ) + await session.commit() + return result + + m = Message( + mailbox_id=mailbox_id, + message_id=message_id, + subject=subject, + from_addr=from_addr, + to_addr=to_addr, + date_header=dt, + received_at=datetime.utcnow(), + raw_eml=raw, + ) + session.add(m) + await session.commit() # id verfügbar, da PG/UUID serverseitig + + result = {"id": str(m.id), "message_id": message_id} + await session.execute( + update(Task) + .where(Task.id == task.id) + .values(status=TaskStatus.FINISHED, result=result, worker_id=self.worker_id) + ) + await session.commit() + + return result + diff --git a/migrations/versions/20260116_create_messages.py b/migrations/versions/20260116_create_messages.py new file mode 100644 index 0000000..95f4106 --- /dev/null +++ b/migrations/versions/20260116_create_messages.py @@ -0,0 +1,28 @@ +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# revision identifiers: +revision = "20260116_create_messages" +down_revision = None # oder deine letzte Revision + +def upgrade(): + op.create_table( + "messages", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("mailbox_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("message_id", sa.String(512), nullable=True), + sa.Column("subject", sa.String(2048), nullable=True), + sa.Column("from_addr", sa.String(2048), nullable=True), + sa.Column("to_addr", sa.String(4096), nullable=True), + sa.Column("date_header", sa.TIMESTAMP(timezone=True), nullable=True), + sa.Column("received_at", sa.TIMESTAMP(timezone=True), nullable=False), + sa.Column("raw_eml", sa.LargeBinary(), nullable=False), + ) + op.create_index("ix_messages_date", "messages", ["date_header"]) + op.create_index("ix_messages_message_id", "messages", ["message_id"]) + op.create_index("ix_messages_mailbox", "messages", ["mailbox_id"]) + +def downgrade(): + op.drop_table("messages") + diff --git a/migrations/versions/918e1dd7c598_merge_message_branch_and_baseline.py b/migrations/versions/918e1dd7c598_merge_message_branch_and_baseline.py new file mode 100644 index 0000000..ef12b43 --- /dev/null +++ b/migrations/versions/918e1dd7c598_merge_message_branch_and_baseline.py @@ -0,0 +1,28 @@ +"""merge message-branch and baseline + +Revision ID: 918e1dd7c598 +Revises: 20260116_create_messages, bb8e2c8e4b8f +Create Date: 2026-01-16 14:20:04.807623 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '918e1dd7c598' +down_revision: Union[str, Sequence[str], None] = ('20260116_create_messages', 'bb8e2c8e4b8f') +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + pass + + +def downgrade() -> None: + """Downgrade schema.""" + pass diff --git a/poetry.lock b/poetry.lock index f985440..c47aa5e 100644 --- a/poetry.lock +++ b/poetry.lock @@ -30,6 +30,21 @@ files = [ pamqp = "3.3.0" yarl = "*" +[[package]] +name = "aiosqlite" +version = "0.22.1" +description = "asyncio bridge to the standard sqlite3 module" +optional = false +python-versions = ">=3.9" +files = [ + {file = "aiosqlite-0.22.1-py3-none-any.whl", hash = "sha256:21c002eb13823fad740196c5a2e9d8e62f6243bd9e7e4a1f87fb5e44ecb4fceb"}, + {file = "aiosqlite-0.22.1.tar.gz", hash = "sha256:043e0bd78d32888c0a9ca90fc788b38796843360c855a7262a532813133a0650"}, +] + +[package.extras] +dev = ["attribution (==1.8.0)", "black (==25.11.0)", "build (>=1.2)", "coverage[toml] (==7.10.7)", "flake8 (==7.3.0)", "flake8-bugbear (==24.12.12)", "flit (==3.12.0)", "mypy (==1.19.0)", "ufmt (==2.8.0)", "usort (==1.0.8.post1)"] +docs = ["sphinx (==8.1.3)", "sphinx-mdinclude (==0.6.2)"] + [[package]] name = "alembic" version = "1.17.2" @@ -232,6 +247,21 @@ idna = ["idna (>=3.10)"] trio = ["trio (>=0.30)"] wmi = ["wmi (>=1.5.1)"] +[[package]] +name = "email-validator" +version = "2.3.0" +description = "A robust email address syntax and deliverability validation library." +optional = false +python-versions = ">=3.8" +files = [ + {file = "email_validator-2.3.0-py3-none-any.whl", hash = "sha256:80f13f623413e6b197ae73bb10bf4eb0908faf509ad8362c5edeb0be7fd450b4"}, + {file = "email_validator-2.3.0.tar.gz", hash = "sha256:9fc05c37f2f6cf439ff414f8fc46d917929974a82244c20eb10231ba60c54426"}, +] + +[package.dependencies] +dnspython = ">=2.0.0" +idna = ">=2.0.0" + [[package]] name = "fastapi" version = "0.115.14" @@ -436,6 +466,17 @@ files = [ [package.extras] all = ["flake8 (>=7.1.1)", "mypy (>=1.11.2)", "pytest (>=8.3.2)", "ruff (>=0.6.2)"] +[[package]] +name = "iniconfig" +version = "2.3.0" +description = "brain-dead simple config-ini parsing" +optional = false +python-versions = ">=3.10" +files = [ + {file = "iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12"}, + {file = "iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730"}, +] + [[package]] name = "mako" version = "1.3.10" @@ -772,6 +813,21 @@ docs = ["furo (>=2025.9.25)", "proselint (>=0.14)", "sphinx (>=8.2.3)", "sphinx- test = ["appdirs (==1.4.4)", "covdefaults (>=2.3)", "pytest (>=8.4.2)", "pytest-cov (>=7)", "pytest-mock (>=3.15.1)"] type = ["mypy (>=1.18.2)"] +[[package]] +name = "pluggy" +version = "1.6.0" +description = "plugin and hook calling mechanisms for python" +optional = false +python-versions = ">=3.9" +files = [ + {file = "pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746"}, + {file = "pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3"}, +] + +[package.extras] +dev = ["pre-commit", "tox"] +testing = ["coverage", "pytest", "pytest-benchmark"] + [[package]] name = "propcache" version = "0.4.1" @@ -916,6 +972,7 @@ files = [ [package.dependencies] annotated-types = ">=0.6.0" +email-validator = {version = ">=2.0.0", optional = true, markers = "extra == \"email\""} pydantic-core = "2.41.5" typing-extensions = ">=4.14.1" typing-inspection = ">=0.4.2" @@ -1057,6 +1114,41 @@ files = [ [package.dependencies] typing-extensions = ">=4.14.1" +[[package]] +name = "pygments" +version = "2.19.2" +description = "Pygments is a syntax highlighting package written in Python." +optional = false +python-versions = ">=3.8" +files = [ + {file = "pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b"}, + {file = "pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887"}, +] + +[package.extras] +windows-terminal = ["colorama (>=0.4.6)"] + +[[package]] +name = "pytest" +version = "9.0.2" +description = "pytest: simple powerful testing with Python" +optional = false +python-versions = ">=3.10" +files = [ + {file = "pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b"}, + {file = "pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11"}, +] + +[package.dependencies] +colorama = {version = ">=0.4", markers = "sys_platform == \"win32\""} +iniconfig = ">=1.0.1" +packaging = ">=22" +pluggy = ">=1.5,<2" +pygments = ">=2.7.2" + +[package.extras] +dev = ["argcomplete", "attrs (>=19.2)", "hypothesis (>=3.56)", "mock", "requests", "setuptools", "xmlschema"] + [[package]] name = "python-dotenv" version = "1.2.1" @@ -1071,6 +1163,17 @@ files = [ [package.extras] cli = ["click (>=5.0)"] +[[package]] +name = "python-multipart" +version = "0.0.21" +description = "A streaming multipart parser for Python" +optional = false +python-versions = ">=3.10" +files = [ + {file = "python_multipart-0.0.21-py3-none-any.whl", hash = "sha256:cf7a6713e01c87aa35387f4774e812c4361150938d20d232800f75ffcf266090"}, + {file = "python_multipart-0.0.21.tar.gz", hash = "sha256:7137ebd4d3bbf70ea1622998f902b97a29434a9e8dc40eb203bbcf7c2a2cba92"}, +] + [[package]] name = "pyyaml" version = "6.0.3" @@ -1198,12 +1301,14 @@ description = "Database Abstraction Library" optional = false python-versions = ">=3.7" files = [ + {file = "sqlalchemy-2.0.45-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:c64772786d9eee72d4d3784c28f0a636af5b0a29f3fe26ff11f55efe90c0bd85"}, {file = "sqlalchemy-2.0.45-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:7ae64ebf7657395824a19bca98ab10eb9a3ecb026bf09524014f1bb81cb598d4"}, {file = "sqlalchemy-2.0.45-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0f02325709d1b1a1489f23a39b318e175a171497374149eae74d612634b234c0"}, {file = "sqlalchemy-2.0.45-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:d2c3684fca8a05f0ac1d9a21c1f4a266983a7ea9180efb80ffeb03861ecd01a0"}, {file = "sqlalchemy-2.0.45-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:040f6f0545b3b7da6b9317fc3e922c9a98fc7243b2a1b39f78390fc0942f7826"}, {file = "sqlalchemy-2.0.45-cp310-cp310-win32.whl", hash = "sha256:830d434d609fe7bfa47c425c445a8b37929f140a7a44cdaf77f6d34df3a7296a"}, {file = "sqlalchemy-2.0.45-cp310-cp310-win_amd64.whl", hash = "sha256:0209d9753671b0da74da2cfbb9ecf9c02f72a759e4b018b3ab35f244c91842c7"}, + {file = "sqlalchemy-2.0.45-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:2e90a344c644a4fa871eb01809c32096487928bd2038bf10f3e4515cb688cc56"}, {file = "sqlalchemy-2.0.45-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b8c8b41b97fba5f62349aa285654230296829672fc9939cd7f35aab246d1c08b"}, {file = "sqlalchemy-2.0.45-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:12c694ed6468333a090d2f60950e4250b928f457e4962389553d6ba5fe9951ac"}, {file = "sqlalchemy-2.0.45-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:f7d27a1d977a1cfef38a0e2e1ca86f09c4212666ce34e6ae542f3ed0a33bc606"}, @@ -1232,12 +1337,14 @@ files = [ {file = "sqlalchemy-2.0.45-cp314-cp314-win_amd64.whl", hash = "sha256:4748601c8ea959e37e03d13dcda4a44837afcd1b21338e637f7c935b8da06177"}, {file = "sqlalchemy-2.0.45-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:cd337d3526ec5298f67d6a30bbbe4ed7e5e68862f0bf6dd21d289f8d37b7d60b"}, {file = "sqlalchemy-2.0.45-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:9a62b446b7d86a3909abbcd1cd3cc550a832f99c2bc37c5b22e1925438b9367b"}, + {file = "sqlalchemy-2.0.45-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:5964f832431b7cdfaaa22a660b4c7eb1dfcd6ed41375f67fd3e3440fd95cb3cc"}, {file = "sqlalchemy-2.0.45-cp38-cp38-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ee580ab50e748208754ae8980cec79ec205983d8cf8b3f7c39067f3d9f2c8e22"}, {file = "sqlalchemy-2.0.45-cp38-cp38-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:13e27397a7810163440c6bfed6b3fe46f1bfb2486eb540315a819abd2c004128"}, {file = "sqlalchemy-2.0.45-cp38-cp38-musllinux_1_2_aarch64.whl", hash = "sha256:ed3635353e55d28e7f4a95c8eda98a5cdc0a0b40b528433fbd41a9ae88f55b3d"}, {file = "sqlalchemy-2.0.45-cp38-cp38-musllinux_1_2_x86_64.whl", hash = "sha256:db6834900338fb13a9123307f0c2cbb1f890a8656fcd5e5448ae3ad5bbe8d312"}, {file = "sqlalchemy-2.0.45-cp38-cp38-win32.whl", hash = "sha256:1d8b4a7a8c9b537509d56d5cd10ecdcfbb95912d72480c8861524efecc6a3fff"}, {file = "sqlalchemy-2.0.45-cp38-cp38-win_amd64.whl", hash = "sha256:ebd300afd2b62679203435f596b2601adafe546cb7282d5a0cd3ed99e423720f"}, + {file = "sqlalchemy-2.0.45-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:d29b2b99d527dbc66dd87c3c3248a5dd789d974a507f4653c969999fc7c1191b"}, {file = "sqlalchemy-2.0.45-cp39-cp39-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:59a8b8bd9c6bedf81ad07c8bd5543eedca55fe9b8780b2b628d495ba55f8db1e"}, {file = "sqlalchemy-2.0.45-cp39-cp39-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:fd93c6f5d65f254ceabe97548c709e073d6da9883343adaa51bf1a913ce93f8e"}, {file = "sqlalchemy-2.0.45-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:6d0beadc2535157070c9c17ecf25ecec31e13c229a8f69196d7590bde8082bf1"}, @@ -1753,4 +1860,4 @@ propcache = ">=0.2.1" [metadata] lock-version = "2.0" python-versions = ">=3.12,<3.13" -content-hash = "11c8982ce9e515f034525f43cb89a6784194964fa2f7b5fe5566178b8153e2d4" +content-hash = "7f07094522b2b192a8d909763d273ccdfa52c0de16fb57fbb063ec9b98e26783" diff --git a/pyproject.toml b/pyproject.toml index 562aecd..db08ad1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,11 +13,14 @@ fastapi = "^0.115" uvicorn = {version = "^0.30", extras = ["standard"]} sqlalchemy = "^2.0" asyncpg = "^0.29" -pydantic = "^2.6" +pydantic = {extras = ["email"], version = "^2.12.5"} aio-pika = "^9.4" httpx = "^0.27" python-dotenv = "^1.0" dnspython = "^2.6" +python-multipart = "^0.0.21" +pytest = "^9.0.2" +aiosqlite = "^0.22.1" [tool.poetry.group.dev.dependencies] black = "^24.10" From 775ed73b317288a80f12ad163873d0feefab7b76 Mon Sep 17 00:00:00 2001 From: Frank Agerholm Date: Fri, 16 Jan 2026 20:55:12 +0100 Subject: [PATCH 2/2] Erste Korrekturen bei der Integration --- app/workers/mail_store_worker.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/app/workers/mail_store_worker.py b/app/workers/mail_store_worker.py index 3de8379..9f554a0 100644 --- a/app/workers/mail_store_worker.py +++ b/app/workers/mail_store_worker.py @@ -14,8 +14,7 @@ from app.models import Message, Task, TaskStatus class MailStoreWorker(BaseWorker): routing_key = "mail.store" - async def handle(self, task) -> dict: - payload = task.job.payload # payload liegt auf dem Job + async def process(self, payload: dict, **meta) -> dict: raw = base64.b64decode(payload["raw_eml_b64"]) mailbox_id = UUID(payload["mailbox_id"]) @@ -68,3 +67,11 @@ class MailStoreWorker(BaseWorker): return result +async def main(): + worker =MailStoreWorker(routing_key("MailSend", "ImportMessage"), worker_id="mail-import-1") + await worker.run() + +if __name__ == "__main__": + import asyncio as _a + _a.run(main()) +