Dj Scaffold

src/
  manage.py
  project/
    __init__.py
    settings.py
    urls.py
    wsgi.py
    asgi.py
    api/
      __init__.py   # NinjaAPI() instance, exception handlers, mounts all resource routers
      <resource>/
        __init__.py  # re-exports router
        routes.py    # handler functions
        schemas.py   # ninja.Schema input types
    types.py        # AuthedRequest and other shared typing aliases
    ids.py          # prefixed ULID generators
    services.py    # svcs registry + get() helper
    signals.py     # ReliableSignal base + send_reliable machinery
  <app>/
    __init__.py
    apps.py
    admin.py
    models/
      __init__.py
      <entity>.py
    dtos/
      __init__.py
      <entity>.py
    repositories/
      __init__.py
      <entity>.py
    services/
      __init__.py
      <entity>.py
    signals.py      # optional, defines ReliableSignal instances
    receivers.py    # optional, @receiver handlers — must be idempotent
tests/
  <app>/
    test_repo.py
    test_service.py
    test_api.py
pyproject.toml

uv add 'django>=6.0' 'django-ninja>=1.6' 'pydantic>=2.0' 'svcs>=25.1' \
       'python-ulid>=3.0' 'celery>=5.4' python-decouple
uv add --dev ruff 'pyrefly>=0.42' django-stubs pytest pytest-django

from ulid import ULID


def prefixed_ulid(prefix: str) -> str:
    return f"{prefix}_{str(ULID()).lower()}"


def _make_generator(prefix: str):
    def generate() -> str:
        return prefixed_ulid(prefix)

    generate.__name__ = f"generate_{prefix}_id"
    generate.__qualname__ = f"generate_{prefix}_id"
    return generate


# Add one generator per aggregate root, with a unique 3-4 char prefix.
# Example:
# generate_prd_id = _make_generator("prd")

import svcs

registry = svcs.Registry()


def register_services(registry: svcs.Registry) -> None:
    """Register every real factory. Tests re-run this to restore overrides."""
    # Register repositories and services here as the project grows.
    # Example:
    # from products.repositories.product import ProductRepository
    # from products.services.product import ProductService
    #
    # registry.register_factory(ProductRepository, ProductRepository)
    #
    # def _product_service_factory(container: svcs.Container) -> ProductService:
    #     return ProductService(container.get(ProductRepository))
    #
    # registry.register_factory(ProductService, _product_service_factory)


register_services(registry)


def get[T](service_type: type[T]) -> T:
    """Get a service from the registry. Works anywhere — views, tasks, commands."""
    return svcs.Container(registry).get(service_type)

# If the project swaps AUTH_USER_MODEL for a custom user (see dj-models),
# import that model here instead — a stale stock-User import makes this
# type silently wrong everywhere it's used.
from django.contrib.auth.models import User
from django.http import HttpRequest


class AuthedRequest(HttpRequest):
    """
    An HttpRequest whose `user` attribute is guaranteed to be an authenticated User.

    Use as the first-argument annotation on any django-ninja handler that requires
    auth. The narrowing is a contract, not runtime enforcement — pair this with
    ninja's `auth=` on the router or a middleware that rejects anonymous requests.
    """
    user: User  # type: ignore[assignment]

from ninja import NinjaAPI

# Import resource routers and mount them below.
# from project.api.products import router as products_router

# TODO: no auth wired — every route is public at this mount point.
# `AuthedRequest` is a typing contract only; it enforces nothing at runtime.
# Wire real auth before shipping, e.g. NinjaAPI(auth=django_auth) or a
# per-router `auth=` argument.
api = NinjaAPI()

# api.add_router("/products", products_router)


@api.exception_handler(ValueError)
def on_value_error(request, exc: ValueError):
    return api.create_response(request, {"detail": str(exc)}, status=400)


@api.exception_handler(LookupError)
def on_lookup_error(request, exc: LookupError):
    return api.create_response(request, {"detail": str(exc)}, status=404)


@api.exception_handler(PermissionError)
def on_permission_error(request, exc: PermissionError):
    return api.create_response(request, {"detail": str(exc)}, status=403)

from typing import List

from ninja import Router

from products.dtos.product import ProductDTO
from products.services.product import ProductService
from project.services import get
from project.types import AuthedRequest

from .schemas import CreateProductIn

router = Router()


@router.get("/", response=List[ProductDTO])
def list_products(request: AuthedRequest):
    return get(ProductService).list_products()

from decimal import Decimal

from ninja import Schema


class CreateProductIn(Schema):
    name: str
    price: Decimal
    stock: int

from .routes import router

__all__ = ["router"]

from django.contrib import admin
from django.urls import path
from project.api import api

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", api.urls),
]

import json

from celery import shared_task
from django.db import transaction
from django.dispatch import Signal
from django.utils.module_loading import import_string


@shared_task
def _dispatch_reliable_receiver(receiver_path: str, kwargs_json: str) -> None:
    receiver = import_string(receiver_path)
    receiver(**json.loads(kwargs_json))


class ReliableSignal(Signal):
    """A Django Signal whose receivers run asynchronously via Celery.

    - `send_reliable()` must be called inside a `transaction.atomic()` block.
    - Receiver tasks are enqueued on transaction commit, so rollbacks are respected.
    - Delivery is at-least-once. Every receiver MUST be idempotent.
    - Arguments MUST be JSON-serializable (pass IDs, never model instances).
    """

    def send_reliable(self, sender, **kwargs) -> None:
        payload = json.dumps(kwargs)
        # _live_receivers() is a private Django API; since Django 5.0 it returns
        # a (sync_receivers, async_receivers) tuple. Re-check on Django upgrades.
        sync_receivers, async_receivers = self._live_receivers(sender)
        for receiver in (*sync_receivers, *async_receivers):
            path = f"{receiver.__module__}.{receiver.__qualname__}"
            transaction.on_commit(
                lambda p=path: _dispatch_reliable_receiver.delay(p, payload)
            )

import os

from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "project.settings")

app = Celery("project")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

from .celery import app as celery_app

__all__ = ("celery_app",)