· 10 min read · Engineering
Advanced Django — Part 3: Selectors and services
This is Part 3 of the Advanced Django series. Part 1 covered the model layer. Part 2 covered custom querysets and managers and made the case that most of the time I don't use them — I use the layer this post is about.
Part 3 covers selectors and services: two function-oriented layers that sit between the ORM and everything else. Selectors are pure reads. Services are writes with atomicity, validation, and side effects. Both are organized as classes of static methods, one per domain — this is the shape I've settled on across every serious Django backend I've written.
The point of splitting them is to keep business logic out of DRF views, out of Celery tasks, out of management commands — and put it in one place that everyone else calls.
Companion repo: the patterns in this series live in
ademolaidowu/django-saas-foundationon GitHub — a working Django + DRF project you can clone, run, and read alongside the posts.
Why split reads from writes
The default Django tutorial shape is: put everything in views. Serializer parses input, view calls Model.objects.create(...), returns a response. That works for a CRUD screen. It falls apart the moment the same logic is needed somewhere that isn't a view — a Celery task processing a webhook, a management command backfilling data, an admin action, a scheduled report.
If placing an order lives inside a DRF perform_create, and now a background job also needs to place an order, you either:
- Duplicate the logic (goodbye single source of truth), or
- Refactor under pressure while the feature is due
The split-early answer: every write is a static method on a service class, every read is a static method on a selector class, and both live in files that don't know anything about HTTP.
Concretely:
- Selectors — read functions grouped in classes like
OrderSelectors. Every method is@staticmethod, keyword-only, no side effects. Returns aQuerySetor a model instance. - Services — write functions grouped in classes like
OrderService. Every method is@staticmethod @transaction.atomic, keyword-only, does one job, returns the object it created or mutated.
Views call them. Celery tasks call them. Management commands call them. Everyone calls the same thing.
Selectors — classes of static methods
I organize selectors as classes named {Entity}Selectors (plural), with every method as @staticmethod. The class doesn't hold state — it's a grouping device that makes import sites and call sites self-documenting.
from orders.selectors import OrderSelectors
order = OrderSelectors.get_order_by_public_id(
order_id=order_id,
organization=organization,
)That OrderSelectors. prefix at every call site tells the reader "this is a read on the Order domain" without them having to look at the function name. It also means every read for a domain sits in one file, easy to grep. Module-level functions would work too — but the class prefix does real navigational work at scale.
The common signature
Every ID-based lookup selector uses the same signature shape. Once you've read one, you've read them all:
# apps/orders/selectors/order.py
from typing import TYPE_CHECKING
from uuid import UUID
from core.exceptions import APIServiceError
from orders import exceptions as order_exceptions
from orders.models import Order
if TYPE_CHECKING:
from tenants.models import Organization
class OrderSelectors:
"""Read-only query functions for the Order domain."""
@staticmethod
def get_order_by_public_id(
*,
order_id: UUID,
organization: "Organization",
select_related: list[str] | None = None,
prefetch_related: list[str] | None = None,
raise_if_missing: bool = True,
) -> Order | None:
qs = Order.records.filter(
public_id=order_id,
organization=organization,
)
if select_related:
qs = qs.select_related(*select_related)
if prefetch_related:
qs = qs.prefetch_related(*prefetch_related)
order = qs.first()
if order is None and raise_if_missing:
raise APIServiceError(error_code=order_exceptions.OrderError0001())
return orderFour things stay consistent across every lookup selector.
Keyword-only arguments. The * in the signature forces every caller to write order_id=..., organization=.... Across module boundaries, positional arguments are a bug waiting to happen. Labels at every call site.
Explicit organization scoping. Every read is scoped to organization (or user) as a required argument. There's no implicit tenant lookup buried inside the selector — the caller passes it, the selector filters by it. This is why I don't need custom manager methods for tenant scoping (Part 2).
select_related and prefetch_related as kwargs. The caller decides what to prefetch, because only the caller knows what fields the response will touch. A detail view asks for select_related=["customer"] and prefetch_related=["line_items"]; a list view might skip both. The selector doesn't guess — it applies what it's given.
raise_if_missing toggle. Two behaviors from one function: raise_if_missing=True (the default) raises APIServiceError with a domain-specific error code when the row doesn't exist; raise_if_missing=False returns None silently for existence checks. One function beats two variants (get_or_none + get_or_raise) because the caller's intent is right there in the argument.
The error-code discipline
APIServiceError(error_code=order_exceptions.OrderError0001()) isn't a plain raise Http404. Each error is a small enum-driven class:
# apps/orders/exceptions.py
class OrderError0001(APIErrorCode):
code = "order_0001"
message = "Order not found"
status = 404The DRF exception handler catches APIServiceError, reads the error_code, and returns a consistent {"error": {"code", "message", "request_id"}} envelope. Enum-driven codes are testable (assert exc.error_code == OrderError0001) and give support a real code to search on. This gets its own post later — for now, the pattern to notice is that selectors and services don't raise Django's DoesNotExist or DRF's NotFound; they raise a domain error.
List selectors and search
List selectors have their own shape. Kwargs stop being about ORM optimization and start being about filters the caller wants applied:
class OrderSelectors:
# ... other methods ...
@staticmethod
def list_orders_for_organization(
*,
organization: "Organization",
search: str | None = None,
status: str | None = None,
active_only: bool = True,
) -> QuerySet[Order]:
qs = Order.records.filter(organization=organization).select_related("customer")
if active_only:
qs = qs.filter(is_active=True)
if status:
qs = qs.filter(status=status)
if search:
qs = qs.filter(
Q(order_number__icontains=search)
| Q(customer__email__icontains=search)
| Q(customer__first_name__icontains=search)
| Q(customer__last_name__icontains=search),
)
return qs.order_by("-created_at")Two conventions here that show up on every list selector:
Search via Q(...) | Q(...) on __icontains. The search kwarg, if present, filters against a handful of user-visible fields with case-insensitive contains. Cheap to write, works fine for lists of tens of thousands of rows on Postgres. If a table grows into the millions and search becomes slow, this is the seam where I'd introduce SearchVector + a GIN index. But I don't add that complexity until the query plan says I need it.
Filter flags as booleans with useful defaults. active_only=True because active is what you want 95% of the time. The caller can flip it off when they need to see everything. Same pattern as other filter flags — include_platform=False, include_suggested=False, etc.
Return a QuerySet, not a list. The view might paginate on top of it. A downstream selector might chain more filters. Returning a QuerySet composes; returning a list forecloses.
Services — classes of atomic static methods
Services follow the same organizational pattern, but each class is named {Entity}Service (singular, because a service class is a namespace for one entity's write operations). Every method is decorated with both @staticmethod and @transaction.atomic, always in that order:
# apps/orders/services/order.py
from typing import TYPE_CHECKING, TypedDict
from uuid import UUID
from decimal import Decimal
from django.db import transaction
from core.exceptions import APIServiceError
from orders import exceptions as order_exceptions
from orders.models import Order, LineItem
from orders.selectors.order import OrderSelectors
if TYPE_CHECKING:
from accounts.models import User
from tenants.models import Organization
class LineItemInput(TypedDict):
product_id: UUID
quantity: int
unit_price: Decimal
class OrderCreateData(TypedDict):
order_number: str
items: list[LineItemInput]
shipping_address: str
class OrderService:
"""Write operations on the Order domain."""
@staticmethod
@transaction.atomic
def create_order(
*,
organization: "Organization",
data: OrderCreateData,
created_by: "User",
) -> Order:
# Idempotency: replayed submissions return the existing row.
existing = Order.records.filter(
organization=organization,
order_number=data["order_number"],
).first()
if existing is not None:
return existing
# Explicit validation, raise before mutating.
if not data["items"]:
raise APIServiceError(error_code=order_exceptions.OrderError0002())
order = Order.records.create(
organization=organization,
order_number=data["order_number"],
shipping_address=data["shipping_address"],
status=Order.Status.PENDING,
created_by=created_by,
)
LineItem.objects.bulk_create([
LineItem(
order=order,
product_id=item["product_id"],
quantity=item["quantity"],
unit_price=item["unit_price"],
)
for item in data["items"]
])
order.total = sum(
item["unit_price"] * item["quantity"] for item in data["items"]
)
order.save(update_fields=["total"])
return orderThings worth naming:
@staticmethod @transaction.atomic on every write. Both decorators, always. The @staticmethod matches the selector pattern (the class is a namespace, not a state holder). The @transaction.atomic means if any part of the service raises, everything rolls back. No half-created orders in the database, ever.
Structured data argument with a TypedDict. Instead of listing 15 keyword arguments on create_order, I take one data: OrderCreateData and let the TypedDict describe its shape. Callers get autocomplete, mypy catches missing keys, and the service signature stays readable. This is the typing discipline from the earlier post applied to real service code.
Idempotency check upfront. Before doing any work, look up whether the row already exists. If it does, return it — a replayed webhook or a double-clicked submit becomes a no-op. This is safer than get_or_create because it's explicit and there's no race with the create path.
Explicit validation, raising domain errors. If items are empty, raise OrderError0002 before mutating anything. Every business rule is a check in the service, not a validator on a serializer and not a clean() method on the model. The rule applies whether the caller is an API, a task, or an admin action — so the check has to live where they all go through, which is the service.
Bulk-create for lists. LineItem.objects.bulk_create([...]) is one INSERT with N rows. Doing it in a for loop with .save() is N INSERTs with N round-trips. On a real order that's the difference between snappy and slow.
The partial-update pattern
Updates need to touch only the fields the caller sent. Otherwise you clobber columns that no one wanted to change, and audit history fills up with meaningless "changed X to X" rows.
The pattern I use everywhere:
class OrderService:
# ... other methods ...
@staticmethod
@transaction.atomic
def update_order(
*,
organization: "Organization",
order_id: UUID,
data: dict,
updated_by: "User",
) -> Order:
order = OrderSelectors.get_order_by_public_id(
order_id=order_id,
organization=organization,
)
update_fields: list[str] = []
scalar_fields = ("shipping_address", "notes", "expected_delivery_at")
for field in scalar_fields:
if field in data:
setattr(order, field, data[field])
update_fields.append(field)
# ... conditional updates for related fields go here ...
if update_fields:
order.updated_by = updated_by
update_fields += ["updated_by", "updated_at"]
order.save(update_fields=update_fields)
return orderThe core idea: build a update_fields list, only append to it when the incoming data actually has that key, then pass the list to save(update_fields=[...]). If nothing changed, don't touch the row at all — the if update_fields guard prevents an empty update.
The audit fields (updated_by, updated_at) get appended only when something else changed. That way audit history reflects real changes, not calls that came in empty.
Related-field updates (foreign keys, M2M) get their own branches — check that the ID resolves to a real row, raise the right error code if it doesn't, then setattr and append. Same shape, more branches.
Idempotency variations
The check_first pattern is my default for services that can be replayed. Two other variants show up when concurrency is in play.
Get-or-create + update for the "the row exists or it doesn't, either way we want the same end state" case:
history, _ = OrderHistory.objects.get_or_create(
order=order,
defaults={"created_by": updated_by},
)
# ...partial update on `history`...IntegrityError race handling when two writers hit a partial unique constraint at the same time:
try:
asset.save()
except IntegrityError:
winner = Asset.objects.filter(
organization=organization,
sha256=sha256,
deleted_at__isnull=True,
).first()
if winner is None:
raise
return winnerThat handles the specific case where you did a "check first, then insert" but a concurrent request tripped the unique on you between the check and the save. Rare but real, and the recovery is: look up the row that won the race, hand it back. The caller doesn't care which write "won".
Views become thin wiring
With selectors and services in place, views have very little left to do:
# apps/orders/views.py
class OrderListView(ListAPIView):
serializer_class = OrderOutput
def get_queryset(self):
return OrderSelectors.list_orders_for_organization(
organization=self.request.organization,
status=self.request.query_params.get("status"),
search=self.request.query_params.get("search"),
)
class OrderDetailView(RetrieveAPIView):
serializer_class = OrderOutput
def get_object(self):
return OrderSelectors.get_order_by_public_id(
order_id=self.kwargs["public_id"],
organization=self.request.organization,
select_related=["customer"],
prefetch_related=["line_items"],
)
class OrderCreateView(CreateAPIView):
def post(self, request):
ser = OrderCreateInput(data=request.data)
ser.is_valid(raise_exception=True)
order = OrderService.create_order(
organization=request.organization,
data=ser.validated_data,
created_by=request.user,
)
return Response(OrderOutput(order).data, status=201)Parse input, call the domain function, serialize output. Every view is that shape. When a view starts doing something else — validation, computation, orchestration — that logic belongs in the service, not the view.
The same service called from a Celery task
The payoff shows up the first time a background job needs the same logic:
# apps/orders/tasks.py
@shared_task
def create_order_from_webhook(*, webhook_data: dict, organization_id: int) -> int:
organization = Organization.records.get(id=organization_id)
user = User.records.get(id=webhook_data["user_id"])
order = OrderService.create_order(
organization=organization,
data=_build_order_data(webhook_data),
created_by=user,
)
return order.idNo duplication. Same validation, same idempotency (the reference check inside create_order means a retried webhook returns the same order), same atomicity.
Testing gets much easier
Tests against selectors and services are plain Python — no HTTP, no request factory, no view instantiation:
@pytest.mark.django_db
def test_create_order_is_idempotent(organization, user):
data = {
"order_number": "ORD-001",
"items": [{"product_id": product.public_id, "quantity": 1, "unit_price": Decimal("10")}],
"shipping_address": "1 High Street",
}
first = OrderService.create_order(organization=organization, data=data, created_by=user)
second = OrderService.create_order(organization=organization, data=data, created_by=user)
assert first.pk == second.pk
assert Order.records.filter(order_number="ORD-001").count() == 1The test asserts a business rule directly. No DRF client, no URL routing, no auth setup. When you need to add a rule ("orders over $10k require a manager approval flag"), you write the test against the service.
What's next in the series
Part 4 covers serializers with an input/output split — how the serializer at the top of a view is different from the serializer at the bottom, why one clever ModelSerializer per model breaks down at scale, and how validation rules end up in exactly one place. That's the last piece of the vertical before Part 5 puts it all together at the view and permission layer.
If you split these layers differently — especially if you've found a shape that beats what's here — I'd genuinely like to hear about it. That's how these posts get better.
- django
- python
- software development