5.1.3. Build helpers
Now we will add code for the helpers in the project. First, we will add the smallest parts, then the larger, more specific ones.
String helpers
Create string.py in helpers/ and add the following code:
def is_empty(value) -> bool:
"""
Kiểm tra xem một chuỗi có rỗng hay không.
Tham số:
value: giá trị muốn kiểm tra.
Trả về:
True nếu là chuỗi rỗng sau khi trim, False nếu không phải chuỗi hoặc không rỗng.
"""
if not isinstance(value, str):
return False
return value.strip() == ""
Number helpers
Create number.py in helpers/ and add the following code:
from typing import Any
def is_actual_number(value: Any) -> bool:
"""
Kiểm tra xem một giá trị có thật sự là số (int hoặc float) trong Python.
Args:
value: Giá trị cần kiểm tra.
Returns:
True nếu là số thực sự (int hoặc float) và không phải NaN, ngược lại False.
"""
return (
isinstance(value, (int, float))
and not isinstance(value, bool)
and not value != value
)
def is_number(value: Any) -> bool:
"""
Kiểm tra xem một giá trị có thể được chuyển đổi thành số hay không.
Args:
value: Giá trị cần kiểm tra (có thể là chuỗi, số, v.v.).
Returns:
True nếu giá trị có thể chuyển thành số (int hoặc float), ngược lại False.
"""
try:
if value != value:
return False
int_val = int(value)
float_val = float(value)
return True
except (ValueError, TypeError):
return False
Check helpers
Create check.py in helpers/.
Import some things first, we will get an import error from "../../core/error", we will solve it when writing the core part.
from typing import Any
from core.error import AppError
# Import other helpers
from .string import is_empty
Next is the function used to check if a string is empty. If it is, throw an error.
def check_empty_or_throw_error(value: Any, value_name: str, msg: str | None = None):
"""
Kiểm tra và có thể ném lỗi nếu như chuỗi là rỗng.
Args:
value: giá trị cần kiểm tra.
value_name: tên của giá trị.
msg: thông báo lỗi tuỳ chỉnh.
Returns:
None
"""
if not msg:
msg = f"{value_name} is empty"
if is_empty(value):
raise AppError(msg)
Next is the function to check if an object is none. If it is, throw an error.
def check_none_or_throw_error(value: Any, value_name: str, msg: str | None = None):
"""
Kiểm tra và có thể ném lỗi nếu như giá trị là undefined.
Args:
value: giá trị cần kiểm tra.
value_name: tên của giá trị.
msg: thông báo lỗi tuỳ chỉnh.
Returns:
None
"""
if not msg:
msg = f"{value_name} is undefined"
if value is None and value_name not in locals():
raise AppError(msg)
Check whether the value of a variable exists, based on two functions check undefined and null.
def check_existance_or_throw_error(value: Any, value_name: str, msg: str | None = None):
"""
Kiểm tra và có thể ném lỗi nếu như giá trị không tồn tại.
Args:
value: giá trị cần kiểm tra.
value_name: tên của giá trị.
msg: thông báo lỗi tuỳ chỉnh.
Returns:
None
"""
check_none_or_throw_error(value, value_name, msg)
Finally, check if a property exists in the object. If not, throw an error.
def check_prop_in_obj_or_throw_error(
obj: Any, obj_name: str, prop_name: str, msg: str | None = None
):
"""
Kiểm tra và có thể ném lỗi nếu như một prop không tồn tại trong obj.
Args:
obj: object cần kiểm tra.
obj_name: tên của object.
prop_name: tên của prop cần kiểm tra.
msg: thông báo lỗi tuỳ chỉnh.
Returns:
None
"""
if not msg:
msg = f"{prop_name} must be in {obj_name}"
if prop_name not in obj or not obj.get(prop_name):
raise AppError(msg)
DynamoDB helpers
Next, create the file dynamodb.py in helpers/.
Import and initialize some things
from typing import Any, Dict, Optional, Callable
# Import external packages
from boto3.dynamodb.types import TypeDeserializer, TypeSerializer
# Import from core
from core.error import ClientError
# Import other helpers
from .number import is_number
serializer = TypeSerializer()
deserializer = TypeDeserializer()
First, we will add the replace_decimals function, it is used to change the type in the DynamoDB Query result so it can be converted to JSON.
def replace_decimals(obj: Any) -> Any:
"""
Thay thế các giá trị kiểu Decimal-like bằng số nguyên hoặc số thực native của Python.
Args:
obj: Đối tượng cần xử lý (có thể là dict, list, hoặc giá trị đơn).
Returns:
Đối tượng đã được chuyển đổi, với các giá trị Decimal-like được thay bằng int hoặc float.
"""
if isinstance(obj, list):
return [replace_decimals(item) for item in obj]
elif isinstance(obj, dict):
return {key: replace_decimals(value) for key, value in obj.items()}
elif hasattr(obj, "to_integral_value"):
num = float(obj)
return int(num) if num.is_integer() else num
return obj
Next is the function to get the value of an attribute in a DynamoDB Item, by the name and type of that attribute.
def get_attr_value(
item: Optional[Dict[str, Any]], attr_name: Optional[str], attr_type: Optional[str]
) -> Any:
"""
Trích xuất giá trị từ một thuộc tính trong DynamoDB item.
Args:
item: Item DynamoDB dưới dạng dict.
attr_name: Tên thuộc tính cần lấy.
attr_type: Kiểu dữ liệu DynamoDB (ví dụ: 'S', 'N', 'BOOL').
Returns:
Giá trị của thuộc tính nếu tồn tại, ngược lại trả về None.
"""
if not item or not attr_name or not attr_type:
return None
attr = item.get(attr_name)
return attr.get(attr_type) if attr else None
Next are 2 conversions from dict to dynamodb item and vice versa.
def to_dynamodb_item(plain: Dict[str, Any]) -> Dict[str, Any]:
"""
Chuyển đổi native Python dict sang DynamoDB item format.
Args:
plain: Đối tượng Python gốc.
Returns:
Đối tượng đã được serialize theo chuẩn DynamoDB.
"""
return {k: serializer.serialize(v) for k, v in plain.items()}
def from_dynamodb_item(dynamo_item: Dict[str, Any]) -> Dict[str, Any]:
"""
Chuyển đổi DynamoDB item sang native Python dict.
Args:
dynamo_item: Item DynamoDB đã được serialize.
Returns:
Đối tượng Python gốc sau khi deserialize.
"""
return {k: deserializer.deserialize(v) for k, v in dynamo_item.items()}
And finally is the function used to build the Update Expression in the input of update item.
def _build_expression_attr_values(
obj: Optional[Dict[str, Any]] = None,
allowed_attrs: Optional[list[str]] = None,
fn: Optional[Callable] = None,
) -> Dict[str, Any]:
"""
Xây dựng ExpressionAttributeValues từ dict đầu vào.
Args:
obj: Dữ liệu đầu vào dạng dict.
allowed_attrs: Danh sách các thuộc tính được phép xử lý.
fn: Hàm callback để xử lý từng key.
Returns:
Dict chứa các ExpressionAttributeValues theo chuẩn DynamoDB.
"""
expression_attr_values: Dict[str, Any] = {}
if not obj:
return expression_attr_values
for key, value in obj.items():
if allowed_attrs and key not in allowed_attrs:
continue
placeholder = f":{key}"
if is_number(value):
expression_attr_values[placeholder] = {"N": str(value)}
elif isinstance(value, str):
expression_attr_values[placeholder] = {"S": value}
elif isinstance(value, list):
expression_attr_values[placeholder] = {
"L": [serializer.serialize(v) for v in value]
}
elif isinstance(value, dict):
expression_attr_values[placeholder] = {
"M": serializer.serialize(value)["M"]
}
elif isinstance(value, bool):
expression_attr_values[placeholder] = {"BOOL": value}
elif isinstance(value, set):
arr = list(value)
type_set = type(arr[0])
if not all(isinstance(v, type_set) for v in arr):
raise ClientError(f"Inconsistent types in set for key {key}")
if type_set is str:
expression_attr_values[placeholder] = {"SS": arr}
elif type_set in [int, float]:
expression_attr_values[placeholder] = {"NS": [str(v) for v in arr]}
if fn:
fn(key, value)
return expression_attr_values
def build_set_update_expression(
obj: Optional[Dict[str, Any]] = None,
allowed_attrs: Optional[list[str]] = None,
) -> Dict[str, Any]:
"""
Tạo biểu thức SET cho DynamoDB update.
Args:
obj: Dữ liệu đầu vào dạng dict.
allowed_attrs: Danh sách các thuộc tính được phép cập nhật.
Returns:
Dict chứa setExpression và expressionAttrValues, hoặc None nếu không có dữ liệu.
"""
if not obj:
return {}
set_parts = []
def collect_expression(key: str, _: Any):
set_parts.append(f"{key} = :{key}")
expression_attr_values = _build_expression_attr_values(
obj, allowed_attrs, collect_expression
)
set_expression = "SET " + ", ".join(set_parts)
return {
"set_expression": set_expression,
"expression_attr_values": expression_attr_values,
}
At this point, we have finished setting up the utils part, in the next part we will add the core part. If you are ready and want to continue, we will proceed with the next part right away.