Skip to content

Part 5: Business Logic Implementation

In the previous part, we defined API endpoints and schema models. Now we will implement the core business logic in the Services layer, connect the endpoints to the database via serializers, and handle edge cases with custom exceptions.

Services Layer

In Unfazed's layered architecture, the Services layer encapsulates business rules and provides data operations to the Endpoints layer:

Endpoints (Controller) → Services (Business Logic) → Serializers / Models (Data)

This separation keeps endpoint functions thin (request parsing and response formatting only) and makes business logic independently testable.

Defining Custom Exceptions

Unfazed's built-in exception hierarchy (see Exceptions) covers common cases like PermissionDenied and MethodNotAllowed. For application-specific errors, subclass BaseUnfazedException:

Create enroll/exceptions.py:

# enroll/exceptions.py

from unfazed.exception import BaseUnfazedException


class NotFound(BaseUnfazedException):
    def __init__(self, message: str = "Resource not found", code: int = 404):
        super().__init__(message, code)


class ValidationError(BaseUnfazedException):
    def __init__(self, message: str = "Validation failed", code: int = 422):
        super().__init__(message, code)

These exceptions carry a message and a numeric code, which can be caught in middleware to build consistent error responses.

Implementing EnrollService

Edit enroll/services.py:

# enroll/services.py

import typing as t

from pydantic import BaseModel

from . import models as m
from . import serializers as s
from .exceptions import NotFound, ValidationError


class EnrollService:

    # --- Request context models ---

    class StudentListQuery(BaseModel):
        name__icontains: str | None = None

    class CourseListQuery(BaseModel):
        is_active: bool | None = None

    class IdCtx(BaseModel):
        id: int

    class CreateStudentCtx(BaseModel):
        name: str
        email: str
        age: int
        student_id: str

    # --- List operations ---

    @classmethod
    async def list_student(
        cls,
        page: int = 1,
        size: int = 10,
        search: str = "",
    ) -> t.Dict:
        cond = {}
        if search:
            cond = {"name__icontains": search}

        result = await s.StudentSerializer.list_from_ctx(
            cond=cond, page=page, size=size
        )
        return {
            "code": 0,
            "message": "success",
            "data": [item.model_dump() for item in result.data],
        }

    @classmethod
    async def list_course(
        cls,
        page: int = 1,
        size: int = 10,
        is_active: bool = True,
    ) -> t.Dict:
        cond = {"is_active": is_active} if is_active else {}

        result = await s.CourseSerializer.list_from_ctx(
            cond=cond, page=page, size=size
        )
        return {
            "code": 0,
            "message": "success",
            "data": [item.model_dump() for item in result.data],
        }

    # --- Detail operations ---

    @classmethod
    async def get_student(cls, student_id: int) -> t.Dict:
        student = await m.Student.get_or_none(id=student_id)
        if not student:
            raise NotFound(f"Student {student_id} does not exist")

        serialized = await s.StudentWithCoursesSerializer.retrieve_from_ctx(
            cls.IdCtx(id=student_id)
        )
        return {
            "code": 0,
            "message": "success",
            "data": serialized.model_dump(),
        }

    # --- Create operations ---

    @classmethod
    async def create_student(cls, data: "EnrollService.CreateStudentCtx") -> t.Dict:
        if await m.Student.get_or_none(student_id=data.student_id):
            raise ValidationError(f"Student ID {data.student_id} already exists")

        if await m.Student.get_or_none(email=data.email):
            raise ValidationError(f"Email {data.email} is already in use")

        student = await s.StudentSerializer.create_from_ctx(data)
        return {
            "code": 0,
            "message": "success",
            "data": student.model_dump(),
        }

    # --- Enrollment operations ---

    @classmethod
    async def bind(cls, student_id: int, course_id: int) -> t.Dict:
        student = await m.Student.get_or_none(id=student_id)
        if not student:
            raise NotFound(f"Student {student_id} does not exist")

        course = await m.Course.get_or_none(id=course_id)
        if not course:
            raise NotFound(f"Course {course_id} does not exist")

        if not course.is_active:
            raise ValidationError(f"Course {course.name} is not active")

        if await student.courses.filter(id=course_id).exists():
            raise ValidationError(
                f"Student {student.name} is already enrolled in {course.name}"
            )

        enrolled_count = await course.students.all().count()
        if enrolled_count >= course.max_students:
            raise ValidationError(f"Course {course.name} is full")

        await student.courses.add(course)

        return {
            "code": 0,
            "message": f"Student {student.name} enrolled in {course.name}",
            "data": {"student_id": student_id, "course_id": course_id},
        }

    @classmethod
    async def unbind(cls, student_id: int, course_id: int) -> t.Dict:
        student = await m.Student.get_or_none(id=student_id)
        if not student:
            raise NotFound(f"Student {student_id} does not exist")

        course = await m.Course.get_or_none(id=course_id)
        if not course:
            raise NotFound(f"Course {course_id} does not exist")

        if not await student.courses.filter(id=course_id).exists():
            raise ValidationError(
                f"Student {student.name} is not enrolled in {course.name}"
            )

        await student.courses.remove(course)

        return {
            "code": 0,
            "message": f"Student {student.name} withdrew from {course.name}",
            "data": {"student_id": student_id, "course_id": course_id},
        }

Key points:

  • list_from_ctx(cond, page, size) returns a Result with .count (total records) and .data (list of serializer instances). See Serializer — list_from_ctx.
  • retrieve_from_ctx(ctx) takes a BaseModel with an id field and returns a serializer instance. See Serializer — retrieve_from_ctx.
  • create_from_ctx(ctx) creates a database record and returns the serialized instance.
  • Custom exceptions (NotFound, ValidationError) are used for business logic errors.

Updating Endpoints

Now update the endpoints to call the service layer:

# enroll/endpoints.py

import typing as t

from unfazed.http import HttpRequest, JsonResponse, PlainTextResponse
from unfazed.route import params as p

from . import schema as s
from .services import EnrollService


async def hello(request: HttpRequest) -> PlainTextResponse:
    """Hello World endpoint"""
    return PlainTextResponse("Hello, World!")


async def list_student(
    request: HttpRequest,
    page: t.Annotated[int, p.Query(default=1, description="Page number", ge=1)],
    size: t.Annotated[int, p.Query(default=10, description="Items per page", ge=1, le=100)],
    search: t.Annotated[str, p.Query(default="", description="Search by name")],
) -> t.Annotated[JsonResponse, p.ResponseSpec(model=s.StudentListResponse)]:
    """Get paginated student list with optional name search"""
    result = await EnrollService.list_student(page, size, search)
    return JsonResponse(result)


async def list_course(
    request: HttpRequest,
    page: t.Annotated[int, p.Query(default=1, description="Page number", ge=1)],
    size: t.Annotated[int, p.Query(default=10, description="Items per page", ge=1, le=100)],
) -> t.Annotated[JsonResponse, p.ResponseSpec(model=s.CourseListResponse)]:
    """Get paginated course list"""
    result = await EnrollService.list_course(page, size)
    return JsonResponse(result)


async def bind(
    request: HttpRequest,
    ctx: t.Annotated[s.BindRequest, p.Json()],
) -> t.Annotated[JsonResponse, p.ResponseSpec(model=s.BindResponse)]:
    """Bind a student to a course"""
    result = await EnrollService.bind(ctx.student_id, ctx.course_id)
    return JsonResponse(result)


async def unbind(
    request: HttpRequest,
    ctx: t.Annotated[s.BindRequest, p.Json()],
) -> t.Annotated[JsonResponse, p.ResponseSpec(model=s.BindResponse)]:
    """Remove a student from a course"""
    result = await EnrollService.unbind(ctx.student_id, ctx.course_id)
    return JsonResponse(result)

Updating Schema

Add the create-student request model to enroll/schema.py:

# Add to enroll/schema.py

class CreateStudentRequest(BaseModel):
    name: str = Field(description="Student name", min_length=1, max_length=100)
    email: str = Field(description="Email address")
    age: int = Field(description="Age", ge=16, le=100)
    student_id: str = Field(description="Student number", min_length=1, max_length=20)

Updating Routes

Add the new endpoints to enroll/routes.py:

# enroll/routes.py

from unfazed.route import path

from .endpoints import hello, list_student, list_course, bind, unbind

patterns = [
    path("/hello", endpoint=hello, name="hello"),

    path("/student-list", endpoint=list_student, name="list_students"),
    path("/course-list", endpoint=list_course, name="list_courses"),

    path("/bind", endpoint=bind, methods=["POST"], name="bind_course"),
    path("/unbind", endpoint=unbind, methods=["POST"], name="unbind_course"),
]

Next Steps

You have implemented complete business logic with database operations and error handling. In the next part, we will:

  • Write comprehensive test cases using Requestfactory
  • Test both the Services layer and the Endpoints layer
  • Use pytest fixtures and parameterized tests

Continue to Part 6: Testing and Quality Assurance.