Domain Services:
When a Rule Doesn't Belong to One Aggregate

Most domain logic fits cleanly inside a single Aggregate Root. But every so often a rule needs to read two independent Aggregates to make a judgment, or it's genuinely unclear which one should own it, or it requires calling an external service an Aggregate has no business doing I/O for. That's the gap a Domain Service fills — and it's worth being precise about, because it's also the pattern most often reached for when it isn't actually needed.

What a Domain Service Is Not

It holds no state. It only holds logic. If a "Domain Service" starts needing state, that's a sign it isn't one — reconsider the design. It also never looks anything up itself:

# wrong — a Domain Service using the Repository directly
class OrderValidationService:
    def __init__(self, order_repository: OrderRepository) -> None:
        self.order_repository = order_repository  # forbidden

    async def validate_order(self, order_id: str) -> bool:
        orders, _ = await self.order_repository.find_orders(order_id=order_id)  # forbidden
        # ...

A Domain Service takes already-loaded domain objects and only judges them. The lookup itself is the Application Service's job, not the Domain Service's.

A Real Example: RefundEligibilityService

The domain rule: a refund requires the original payment to be in the COMPLETED state, and the refund amount can't exceed the payment amount. The Payment Aggregate doesn't know about any refund attempt against it — a refund only ever exists as a separate Aggregate. The Refund Aggregate doesn't know the original payment's amount or status either — it only references it via paymentId.

Putting this judgment inside either Aggregate's own method would mean that Aggregate has to take the entire other Aggregate as a parameter, which breaks the boundary both of them are supposed to protect. So the judgment lives in a Domain Service that the Application layer — having loaded both Aggregates independently — delegates to:

# domain/refund_eligibility_service.py — a Domain Service (no framework dependency)
@dataclass(frozen=True)
class RefundDecision:
    approved: bool
    reason: str | None = None


class RefundEligibilityService:
    def evaluate(
        self,
        payment: Payment,
        refund: Refund,
        classification: RefundReasonClassification,
        ml_fraud_risk_score: float,
    ) -> RefundDecision:
        if payment.status != PaymentStatus.COMPLETED:
            return RefundDecision(approved=False, reason="A refund can only be requested for a completed payment.")
        if refund.amount > payment.amount:
            return RefundDecision(approved=False, reason="The refund amount cannot exceed the payment amount.")
        if (
            classification.category == RefundReasonCategory.FRAUD_SUSPECTED
            and classification.fraud_risk_score >= FRAUD_RISK_REJECTION_THRESHOLD
        ):
            return RefundDecision(approved=False, reason="This refund reason was flagged as high fraud risk and requires manual review.")
        if ml_fraud_risk_score >= ML_FRAUD_RISK_REJECTION_THRESHOLD:
            return RefundDecision(approved=False, reason="This refund pattern was flagged as high risk by the fraud-risk model and requires manual review.")
        return RefundDecision(approved=True)

# application/command/request_refund_handler.py — loads both Repositories, classifies the reason,
# scores the history pattern, and delegates all four inputs to the Domain Service
class RequestRefundHandler:
    def __init__(
        self,
        payment_repo: PaymentRepository,
        refund_repo: RefundRepository,
        refund_reason_classifier: RefundReasonClassifier,
        refund_fraud_risk_scorer: RefundFraudRiskScorer,
    ) -> None:
        self._payment_repo = payment_repo
        self._refund_repo = refund_repo
        self._refund_reason_classifier = refund_reason_classifier
        self._refund_fraud_risk_scorer = refund_fraud_risk_scorer
        # RefundEligibilityService is a pure Domain Service with no framework dependency —
        # instantiated directly rather than registered with FastAPI's Depends().
        self._refund_eligibility_service = RefundEligibilityService()

    async def execute(self, cmd: RequestRefundCommand) -> Refund:
        payments, _ = await self._payment_repo.find_payments(
            page=0, take=1, payment_id=cmd.payment_id, owner_id=cmd.requester_id
        )
        payment = payments[0] if payments else None
        if payment is None:
            raise PaymentNotFoundError(cmd.payment_id)

        refund = Refund.create(payment_id=payment.payment_id, amount=cmd.amount, reason=cmd.reason)
        classification = await self._refund_reason_classifier.classify(cmd.reason)
        ml_fraud_risk_score = await self._refund_fraud_risk_scorer.score(/* refund history features */)

        decision = self._refund_eligibility_service.evaluate(payment, refund, classification, ml_fraud_risk_score)
        if decision.approved:
            refund.approve(account_id=payment.account_id, owner_id=payment.owner_id)
        else:
            refund.reject(decision.reason or "The refund request was rejected.")

        await self._refund_repo.save_refund(refund)
        return refund

RefundEligibilityService is instantiated directly — it's never wired through FastAPI's Depends(), staying true to "holds no state, no framework dependency." Its unit test doesn't go through the Application layer at all; it instantiates the class directly and verifies only the decision logic. classification and ml_fraud_risk_score are two independent signals produced by Technical Services upstream (an LLM-backed classifier and a history-scoring model, each covered in its own post) — this Domain Service never calls either one, only weighs the already-computed values against its own fixed thresholds. The full code lives at implementations/fastapi/examples/src/payment/domain/refund_eligibility_service.py alongside payment.py, refund.py, and the command handler above.

This example also earned itself a permanent regression guard: a harness rule checks, within payment/domain/, that payment.py never directly imports the Refund class and vice versa — proving the two Aggregates only ever reference each other by ID, never by holding one another as a field. The legitimate pattern of a Domain Service taking both as function parameters, like evaluate(payment: Payment, refund: Refund), is explicitly not a target of that rule.

Domain Service vs. Application Service vs. Technical Service

Three easily-confused concepts, told apart by what they depend on. The Application Service coordinates the use case — calling the Repository, running the transaction. The Domain Service handles the domain judgment inside it, depending only on other domain objects. The Technical Service handles the piece where a technical implementation is the actual point — encryption, file storage, an external API client — abstracted behind an interface the Application layer depends on, with the real SDK usage confined to the Infrastructure-layer implementation.

The difference from a Technical Service is worth spelling out, since both get injected into an Application Service and both look like "just another dependency" from the call site. A Technical Service's interface is shaped around a technical concern unrelated to any domain rule — CryptoService.encrypt() doesn't know what an order or a payment is. A Domain Service's interface is shaped around a domain judgment — RefundEligibilityService.evaluate() is meaningless outside the Payment/Refund domain.

When the Rule Really Doesn't Need a Domain Service

Not every calculation involving a value object is evidence of cross-Aggregate coordination. A discount calculation that only touches one Order and a plain coupon value object is really just Aggregate logic that happens to live in a helper class — it doesn't need two independently-loaded Aggregates to make its decision. The tell for a genuine Domain Service is that the Application layer has to load two separate Repositories and hand both results to something, because neither Aggregate alone has enough information to decide.

Getting this distinction wrong in either direction costs you something concrete: forcing the rule into one Aggregate means that Aggregate now imports and understands a class it has no business depending on; splitting out a Domain Service for logic that only ever touches one Aggregate just adds an indirection with nothing to show for it.