I just completed a (too) long debugging session trying to figure out why background jobs "weren't running" in CoopWise. The real story: FastAPI and Celery are two separate OS processes importing the same in my Coopwise codebase, and anything that relies on an import-time side effect to get initialized; this could be module-level singletons, os.environ mutation, SQLAlchemy's lazy mapper registry — silently diverges between them.
I caught four major bug in this same same family:
1. Circular import that only broke under Celery's entrypoint. Same code, different import order:
# kyc/tasks.py — top-level import triggered a chain back into itself
# under Celery's include=[...], but not under uvicorn (which imports
# src.api first, so the chain resolves before ever revisiting this file)
from src.domains.kyc.dependencies import build_kyc_service
Fixed by deferring the import into the task body:
@celery_app.task(bind=True, max_retries=3, default_retry_delay=30)
def process_identity_submission(self, *, kyc_id, ...):
from src.domains.kyc.dependencies import build_kyc_service
...
2. load_dotenv() in main.py was leaking Celery config through os.environ.
CELERY_BROKER_URL=redis://localhost:6379/1
CELERY_RESULT_BACKEND=redis://localhost:6379/2
load_dotenv() writes .env into os.environ for the whole process. Celery auto-detects CELERY_BROKER_URL/CELERY_RESULT_BACKEND as env overrides(note that as the current scale of Coopwise, I decided to use just one redis instance/url for both the broker and the backend)— so FastAPI's process silently repointed the shared celery_app singleton at different Redis DBs than the one the worker (started via CLI, no load_dotenv()) was listening on. No exception anywhere. Tasks just vanished into a DB nobody read. Deleted the call, deleted the dead config keys. Now secrets are only read and validated in config.py.
3. db_manager only got initialized inside FastAPI's lifespan. The worker imported the same singleton but never ran its init, so every task hit RuntimeError("Database not initialized").
4. SQLAlchemy's mapper registry only knows about models actually imported in-process. FastAPI's router chain imports everything by request time; the worker's narrow task-only imports didn't, so a relationship("CooperativeGroup") string failed to resolve the first time a query actually ran.
The fix, centralized into one place instead of scattered per task file:
# src/infra/celery/app.py
@worker_process_init.connect
def _bootstrap_worker_process(**kwargs) -> None:
"""Everything a Celery worker child needs that FastAPI's lifespan
would otherwise provide for free. Add to this, don't scatter
equivalent init logic across individual task files."""
import sqlalchemy
from config import AppConfig as config
from src.infra.db.database import db_manager
import src.infra.db.all_models # noqa: F401 — populate Base.registry
engine_kwargs = {}
if "sqlite" in config.DATABASE_URL:
engine_kwargs.update({
"connect_args": {"check_same_thread": False},
"poolclass": sqlalchemy.StaticPool,
})
db_manager.initialize(
config.DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://", 1),
**engine_kwargs,
)
logger.info("[worker_process_init] worker process bootstrap complete")
The pattern: anything that looks like "call .initialize() once at startup" is actually "call it once per process." Two entrypoints sharing a codebase but not a startup routine means every one of those singletons needs its own explicit init path.
I've hit a version of this before — a Node/BullMQ app where the worker needed to run as a genuinely separate pm2 app via ecosystem.config.js, in a codebase I contribute to rather than own. PR's up for review on that today. Same root cause, different ecosystem: a worker process that doesn't inherit whatever the main app's entrypoint does implicitly.
Also confirms something I already suspected — this would've surfaced faster, and been easier to reason about, if FastAPI and Celery were separate Docker services from the start instead of two backgrounded processes in one dev script. Separate containers force you to be explicit about what each process actually initializes, instead of relying on "well it worked when I ran the API" as a signal that anything is wired correctly.
What I also cleaned up along the way: replaced a fork-conditional logging patch (which Celery's own logging hijack was quietly overwriting) with after_setup_logger/after_setup_task_logger signals, added a celery inspect ping liveness check to the dev script instead of assuming a backgrounded process came up clean, and turned a silent idempotency-lock no-op into a logged skip. Most of this wasn't hard once it was actually observable — the real work was making the failure modes loud instead of silent.