5.5.2. Build routes

Now we already have the code that implements the main features, but how will the client use it? This is where the concept of API and Endpoints comes in. In this section, we will set up the Routes, which will be integrated together in the final step.

Auth Routes

First, in the routes/auth folder, create the file main.py and add some code.

Import a few necessary items first.

from typing import Annotated, Dict

from fastapi import APIRouter, Request, Response, Body

# Import pipelines
from core.modules.auth.ports.main import sign_in_pipeline, refresh_tokens_pipeline

# Import schema
from core.docs.swagger.helpers import StandardJsonResponse
from core.modules.auth.data_model.schema import (
    SignInDataDecriptiveObject,
    SignInResultSchema,
    RefreshTokensDescriptiveObject,
    RefreshTokensResulSchema,
)

# Import from runtimes
from runtimes.fastapi.adapters.context import FastAPIRuntimeContext

Now we’ll set up the routes according to the OpenAPI standard. In auth, we only need to expose two functions that are set up with pipelines.

router = APIRouter()
route_tag = "Auth"


@router.post(
    "/auth/sign-in",
    tags=[route_tag],
    response_model=StandardJsonResponse[SignInResultSchema, Dict],
)
async def handle_sign_in(
    req: Request,
    res: Response,
    _: Annotated[dict, Body(examples=[SignInDataDecriptiveObject])],
):
    ctx = FastAPIRuntimeContext(req, res)
    return await sign_in_pipeline.run(ctx)


@router.post(
    "/auth/refresh-tokens",
    tags=[route_tag],
    response_model=StandardJsonResponse[RefreshTokensResulSchema, Dict],
)
async def handle_refresh_tokens(
    req: Request,
    res: Response,
    _: Annotated[dict, Body(examples=[RefreshTokensDescriptiveObject])],
):
    ctx = FastAPIRuntimeContext(req, res)
    return await refresh_tokens_pipeline.run(ctx)

5.5.2.1

You can see that when a request is sent, an instance of the runtime context is created, and then its pipeline runs the run() function and returns the result from it. This is how we use the runtime context with the pipeline.

Potential Customer Management Routes

The routes for the customer management module are similar, but there will be more routes.

Similarly, create main.py in the pcustomer_management folder and import some necessary items first.

from typing import Annotated, List, Dict

from fastapi import APIRouter, Request, Response, Body, Query, Path

# Import pipelines
from core.modules.pcustomer_management.ports.main import (
    get_customer_pipeline,
    get_customers_pipeline,
    add_customer_pipeline,
    update_customer_pipeline,
    delete_customer_pipeline,
)

# Import schema
from core.docs.swagger.helpers import StandardJsonResponse
from core.modules.pcustomer_management.data_model.schema import (
    PCustomerSchema,
    GetPCustomersResultMetaSchema,
    CreatePCustomerSchema,
    CreatePCustomerDescriptiveObject,
    UpdatePCustomerSchema,
    UpdatePCustomerDescriptiveObject,
)

# Import from runtimes
from runtimes.fastapi.adapters.context import FastAPIRuntimeContext

Then set up the routes according to the OpenAPI standard.

router = APIRouter()
route_tag = "Potential Customers"


@router.get(
    "/pcustomers",
    tags=[route_tag],
    response_model=StandardJsonResponse[
        List[PCustomerSchema], GetPCustomersResultMetaSchema
    ],
)
async def handle_get_pcustomers(
    req: Request,
    res: Response,
    limit: Annotated[str, Query(examples=["10"])] = "",
    startKey: Annotated[str, Query(examples=["base64_string"])] = "",
):
    ctx = FastAPIRuntimeContext(req, res)
    return await get_customers_pipeline.run(ctx)


@router.get(
    "/pcustomers/{id}",
    tags=[route_tag],
    response_model=StandardJsonResponse[PCustomerSchema, Dict],
)
async def handle_get_pcustomer(
    req: Request,
    res: Response,
    id: Annotated[str, Path(examples=["CUSTOMER#___"])],
):
    ctx = FastAPIRuntimeContext(req, res)
    return await get_customer_pipeline.run(ctx)


@router.post(
    "/pcustomer",
    tags=[route_tag],
    response_model=StandardJsonResponse[PCustomerSchema, Dict],
)
async def handle_add_pcustomer(
    req: Request,
    res: Response,
    _: Annotated[Dict, Body(examples=[CreatePCustomerDescriptiveObject])],
):
    ctx = FastAPIRuntimeContext(req, res)
    return await add_customer_pipeline.run(ctx)


@router.patch(
    "/pcustomers/{id}",
    tags=[route_tag],
    response_model=StandardJsonResponse[PCustomerSchema, Dict],
)
async def handle_update_pcustomer(
    req: Request,
    res: Response,
    id: Annotated[str, Path(examples=["CUSTOMER#___"])],
    _: Annotated[Dict, Body(examples=[UpdatePCustomerDescriptiveObject])],
):
    ctx = FastAPIRuntimeContext(req, res)
    return await update_customer_pipeline.run(ctx)


@router.delete(
    "/pcustomers/{id}",
    tags=[route_tag],
    response_model=StandardJsonResponse[bool, Dict],
)
async def handle_delete_pcustomer(
    req: Request,
    res: Response,
    id: Annotated[str, Path(examples=["CUSTOMER#___"])],
):
    ctx = FastAPIRuntimeContext(req, res)
    return await delete_customer_pipeline.run(ctx)

5.5.2.2

5.5.2.3