5.3.2. Build functions
Pre-setup
Before building the features, first create a new folder named functions inside modules/pcustomer_management. Inside the functions folder, create the following files: add_customer.py, delete_customer.py, get_pcustomer.py, get_pcustomers.py, and update_customer.py.
Add Potential Customer Function
First, we build the feature to add a user. Start by importing the necessary modules:
# Import from core
from core.context.runtime_context import RuntimeContext
from core.context.internal_context.main import initialize_internal_context
from core.error import AppError, is_standard_error
from core.modules.pcustomer_management.data_model.dao import PCustomerDAO
Next, the function to add a user:
async def add_customer(ctx: RuntimeContext):
"""
Thêm thông tin của một khách hàng vào trong hệ thống.
Args:
ctx (RuntimeContext): Ngữ cảnh runtime chứa thông tin request.
Returns:
dict | AppError: Thông tin khách hàng vừa thêm hoặc lỗi chuẩn hóa.
"""
try:
body = await ctx.get_body()
pcustomer_dao = PCustomerDAO()
internal_ctx = initialize_internal_context()
internal_ctx.params = body
internal_ctx.options["can_catch_error"] = True
result = await pcustomer_dao.insert_pcustomer(internal_ctx)
return result
except Exception as error:
if is_standard_error(error):
return error
err = AppError("Cannot add new potential customer")
err.as_http_error("InternalServerError")
return err
In this function, you can see that we retrieve user information using ctx.get_body() (Runtime Context) and create another context (Internal Context). Then, we update the information in the internal context and call pcustomerDao.insert_pcustomer(). The following functions operate in a similar flow.
We can still unit test this function, provided the context is fully prepared.
Delete Potential Customer Function
Next, the feature to delete a customer. First, import some modules:
# Import from core
from core.context.runtime_context import RuntimeContext
from core.context.internal_context.main import initialize_internal_context
from core.error import AppError, is_standard_error
from core.modules.pcustomer_management.data_model.dao import PCustomerDAO
Add the function code:
async def delete_customer(ctx: RuntimeContext):
"""
Xoá thông tin của một khách hàng ra khỏi hệ thống.
Args:
ctx (RuntimeContext): Ngữ cảnh runtime chứa thông tin request.
Returns:
bool | AppError: Thông tin khách hàng vừa thêm hoặc lỗi chuẩn hóa.
"""
try:
params = ctx.get_params()
pcustomer_dao = PCustomerDAO()
internal_ctx = initialize_internal_context()
internal_ctx.params = {"query": {"id": params.get("id")}}
internal_ctx.options["can_catch_error"] = True
await pcustomer_dao.delete_pcustomer(internal_ctx)
return True
except Exception as error:
if is_standard_error(error):
return error
err = AppError("Cannot delete potential customer")
err.as_http_error("InternalServerError")
return err
Get Potential Customer Function
Next, the feature to retrieve a customer’s information. First, import some modules:
# Import from core
from core.context.runtime_context import RuntimeContext
from core.context.internal_context.main import initialize_internal_context
from core.error import AppError, is_standard_error
from core.modules.pcustomer_management.data_model.dao import PCustomerDAO
Add the function code:
async def get_pcustomer(ctx: RuntimeContext):
"""
Lấy thông tin của một khách hàng ở trong hệ thống.
Args:
ctx (RuntimeContext): Ngữ cảnh runtime chứa thông tin request.
Returns:
dict | AppError: Thông tin khách hàng hoặc lỗi chuẩn hóa.
"""
try:
params = ctx.get_params()
pcustomer_dao = PCustomerDAO()
internal_ctx = initialize_internal_context()
internal_ctx.params = {"query": {"id": params.get("id")}}
internal_ctx.options["can_catch_error"] = True
result = await pcustomer_dao.get_pcustomer(internal_ctx)
return result
except Exception as error:
if is_standard_error(error):
return error
err = AppError("Cannot get potential customer")
err.as_http_error("InternalServerError")
return err
Get Potential Customers Function
Next, the feature to retrieve a list of customers. First, import some modules:
# Import from core
from core.context.runtime_context import RuntimeContext
from core.context.internal_context.main import initialize_internal_context
from core.error import AppError, is_standard_error
from core.modules.pcustomer_management.data_model.dao import PCustomerDAO
Add the function code:
async def get_pcustomers(ctx: RuntimeContext):
"""
Lấy thông tin của nhiều khách hàng ở trong hệ thống.
Args:
ctx (RuntimeContext): Ngữ cảnh runtime chứa thông tin request.
Returns:
dict | AppError: Danh sách thông tin khách hàng hoặc lỗi chuẩn hóa.
"""
try:
query = ctx.get_query()
pcustomer_dao = PCustomerDAO()
internal_ctx = initialize_internal_context()
internal_ctx.params = {
"limit": query.get("limit", "10"),
"start_key": query.get("startKey"),
}
internal_ctx.options["can_catch_error"] = True
result = await pcustomer_dao.list_pcustomers(internal_ctx)
return result
except Exception as error:
if is_standard_error(error):
return error
err = AppError("Cannot find list of potential customers.")
err.as_http_error("InternalServerError")
return err
Update Potential Customer Function
Next, the feature to update customer information. First, import some modules:
# Import from core
from core.context.runtime_context import RuntimeContext
from core.context.internal_context.main import initialize_internal_context
from core.error import AppError, is_standard_error
from core.modules.pcustomer_management.data_model.dao import PCustomerDAO
Add the function code:
async def update_customer(ctx: RuntimeContext):
"""
Cập nhật thông tin của một khách hàng trong hệ thống.
Args:
ctx (RuntimeContext): Ngữ cảnh runtime chứa thông tin request.
Returns:
dict | AppError: Thông tin khách hàng vừa cập nhật hoặc lỗi chuẩn hóa.
"""
try:
body = await ctx.get_body()
params = ctx.get_params()
pcustomer_dao = PCustomerDAO()
internal_ctx = initialize_internal_context()
internal_ctx.params = {"id": params.get("id"), **body}
internal_ctx.options["can_catch_error"] = True
result = await pcustomer_dao.update_pcustomer(internal_ctx)
return result
except Exception as error:
if is_standard_error(error):
return error
err = AppError("Cannot update existing potential customer")
err.as_http_error("InternalServerError")
return err
With this, we have completed building the functions for managing customers.
Finally, create an __init__.py file to export all these functions from a single entry point (the folder).
from .add_customer import add_customer
from .delete_customer import delete_customer
from .get_pcustomer import get_pcustomer
from .get_pcustomers import get_pcustomers
from .update_customer import update_customer
__all__ = [
"add_customer",
"delete_customer",
"get_pcustomer",
"get_pcustomers",
"update_customer",
]
