5.3.3. Build ports
Now, if we want these functions to be fully featured, we need to use the pipeline. As mentioned earlier, we aim to build the core code into functions so that they can be embedded in pipeline steps. When the pipeline runs, these steps will execute automatically.
Pre-setup
We need to create a ports folder inside modules/pcustomer_management. Inside this folder, create index.ts and add some code as follows.
First, import the necessary modules:
from core.context.pipeline import Pipeline
# Import constants
from core.modules.auth.constants import TEAMS, ROLES
from core.error import AppError, is_standard_error
from core.modules.auth.functions import (
create_roles_check_step_executor,
create_teams_check_step_executor,
create_verify_token_step_executor,
)
# Import functions
from core.modules.pcustomer_management.functions import (
get_pcustomer,
get_pcustomers,
add_customer,
update_customer,
delete_customer,
)
Next, initialize the pipelines as follows:
get_customer_pipeline = Pipeline("Get Customer Pipeline")
get_customers_pipeline = Pipeline("Get Customers Pipeline")
add_customer_pipeline = Pipeline("Add Customer Pipeline")
update_customer_pipeline = Pipeline("Update Customer Pipeline")
delete_customer_pipeline = Pipeline("Delete Customer Pipeline")
When a pipeline runs, it follows these steps:
- Check authorization with
create_verify_token_step_executor(). - Check the requester’s role (to verify if they can proceed).
- Check the requester’s team (to verify if they can proceed).
- Execute the function.
- Check the final result: if no error, return JSON; if there is an error, return the Error.
There may be some errors at this stage, but they will disappear once the auth module is built.
Get Potential Customer Pipeline
First, we build a pipeline to retrieve a customer’s information. This pipeline requires the requester to have the role ADMIN or EMPLOYEE and belong to the MARKETING or SALES team.
get_customer_pipeline.add_step(
create_verify_token_step_executor(get_customer_pipeline)
).add_step(
create_roles_check_step_executor(
get_customer_pipeline, [ROLES["EMPLOYEE"]["NAME"], ROLES["ADMIN"]["NAME"]]
)
).add_step(
create_teams_check_step_executor(
get_customer_pipeline, [TEAMS["MARKETING"]["NAME"], TEAMS["SALES"]["NAME"]]
)
).add_step(
get_pcustomer
).add_step(
lambda ctx: (
ctx.send_error(ctx.prev_result)
if is_standard_error(ctx.prev_result)
else ctx.send_json(ctx.prev_result)
)
)
Get Potential Customers Pipeline
Next, we build a pipeline to retrieve a list of customers. This pipeline also requires the requester to have the role ADMIN or EMPLOYEE and belong to the MARKETING or SALES team.
get_customers_pipeline.add_step(
create_verify_token_step_executor(get_customers_pipeline)
).add_step(
create_roles_check_step_executor(
get_customers_pipeline, [ROLES["EMPLOYEE"]["NAME"], ROLES["ADMIN"]["NAME"]]
)
).add_step(
create_teams_check_step_executor(
get_customers_pipeline, [TEAMS["MARKETING"]["NAME"], TEAMS["SALES"]["NAME"]]
)
).add_step(
get_pcustomers
).add_step(
lambda ctx: (
ctx.send_error(ctx.prev_result)
if is_standard_error(ctx.prev_result)
else ctx.send_json(ctx.prev_result)
)
)
Add Potential Customer Pipeline
Then, we build a pipeline to add customer information. This pipeline requires the requester to have the role ADMIN or EMPLOYEE and belong to the SALES team.
add_customer_pipeline.add_step(
create_verify_token_step_executor(add_customer_pipeline)
).add_step(
create_roles_check_step_executor(
add_customer_pipeline, [ROLES["EMPLOYEE"]["NAME"], ROLES["ADMIN"]["NAME"]]
)
).add_step(
create_teams_check_step_executor(add_customer_pipeline, [TEAMS["SALES"]["NAME"]])
).add_step(
add_customer
).add_step(
lambda ctx: (
ctx.send_error(ctx.prev_result)
if is_standard_error(ctx.prev_result)
else ctx.send_json(ctx.prev_result)
)
)
Update Potential Customer Pipeline
Next, we build a pipeline to update customer information. This pipeline requires the requester to have the role ADMIN or EMPLOYEE and belong to the SALES team.
update_customer_pipeline.add_step(
create_verify_token_step_executor(update_customer_pipeline)
).add_step(
create_roles_check_step_executor(
update_customer_pipeline, [ROLES["EMPLOYEE"]["NAME"], ROLES["ADMIN"]["NAME"]]
)
).add_step(
create_teams_check_step_executor(update_customer_pipeline, [TEAMS["SALES"]["NAME"]])
).add_step(
update_customer
).add_step(
lambda ctx: (
ctx.send_error(ctx.prev_result)
if is_standard_error(ctx.prev_result)
else ctx.send_json(ctx.prev_result)
)
)
Delete Potential Customer Pipeline
Finally, we build a pipeline to delete customer information. This pipeline requires the requester to have the role ADMIN or EMPLOYEE and belong to the SALES team.
delete_customer_pipeline.add_step(
create_verify_token_step_executor(delete_customer_pipeline)
).add_step(
create_roles_check_step_executor(
delete_customer_pipeline, [ROLES["EMPLOYEE"]["NAME"], ROLES["ADMIN"]["NAME"]]
)
).add_step(
create_teams_check_step_executor(delete_customer_pipeline, [TEAMS["SALES"]["NAME"]])
).add_step(
delete_customer
).add_step(
lambda ctx: (
ctx.send_error(ctx.prev_result)
if is_standard_error(ctx.prev_result)
else ctx.send_json(ctx.prev_result)
)
)
If the role or team checks fail, the pipeline stops immediately. In the next section, when we build the auth module, you will get a clearer understanding of the functions create_roles_check_step_executor and create_teams_check_step_executor.
