import dataclasses from typing import Optional import typing import crud from flask import jsonify, request from helpers import CORS_HEADERS, db_rows_to_array, kratos, str_to_bool from pydantic import BaseModel, Field, ValidationError from helpers import kratos, init_db_connection from schemas import UserEmailRequest @dataclasses.dataclass # Filter user bao nhieu email class EmailFilter: ids: typing.Optional[typing.List[str]] = None email: typing.Optional[str] = None provider: typing.Optional[str] = None created_from: typing.Optional[str] = None created_to: typing.Optional[str] = None modified_from: typing.Optional[str] = None modified_to: typing.Optional[str] = None keywords: typing.Optional[str] = None primary: typing.Optional[bool] = None @classmethod def from_request_queries(cls) -> "EmailFilter": return cls( ids=request.args.getlist("filter[ids]"), email=request.args.get("filter[email]"), provider=request.args.get("filter[provider]"), created_from=request.args.get("filter[created_from]"), created_to=request.args.get("filter[created_to]"), modified_from=request.args.get("filter[modified_from]"), modified_to=request.args.get("filter[modified_to]"), keywords=request.args.get("filter[key]"), primary=str_to_bool(request.args.get("filter[primary]")) ) @dataclasses.dataclass class Page: page: int = 0 size: int = 10 asc: bool = False @classmethod def from_request_queries(cls) -> "Page": # Phan Trang Email return Page( page=int(request.args.get("page", 0)), # Neu Client khong truyen Page maic dinh la page 0 (trang 1) va size 10(10 phan tu) size=int(request.args.get("size", 10)), asc=request.args.get("asc", type=str_to_bool) or False ) @dataclasses.dataclass class EmailPage(Page): # Ke thua Page sortby: typing.Optional[str] = None filter: EmailFilter = dataclasses.field( default_factory=EmailFilter.from_request_queries) # Dung qua object field filter theo dang composition # Khi tạo EmailPage mà không truyền sẵn filter, Python sẽ tự gọi EmailFilter.from_request_queries() # Nếu khi tạo object mà field này không được truyền vào, thì hãy gọi HÀM NÀY để tạo giá trị mặc định cho field đó # = dataclasses.field(...) → cấu hình thêm: => Dùng default_factory để tự tạo giá trị nếu bạn không truyền. @classmethod def from_request_queries(cls) -> "EmailPage": base = Page.from_request_queries() # lấy page, size, asc từ URL return cls(**dataclasses.asdict(base), sortby=request.args.get("sortby")) # ở đây không truyền filter -> default_factory được kích hoạt def main(): """ ```fission { "name": "email-users-insert-get-filter", "http_triggers": { "email-users-insert-get-filter-http": { "url": "/ailbl/users/emails", "methods": ["POST", "GET"] } } } ``` """ try: if request.method == "POST": return user_insert_email() elif request.method == "GET": return user_filter_emails() else: return {"error": "Method not allow"}, 405 except Exception as ex: return jsonify({"error": str(ex)}), 500 def user_insert_email(): # 1. Lấy user_id từ header (identity.id Kratos) user_id = request.headers.get("X-UserId") if not user_id: return jsonify({"errorCode": "USER_ID_REQUIRED"}), 400, CORS_HEADERS try: data = request.get_json() if not data: return jsonify({"errorCode": "NO_DATA_PROVIDED"}), 400, CORS_HEADERS parsed = UserEmailRequest(**data) # parsed(object) luu du lieu validate body ma data gui len la 1 dict except ValidationError as e: return jsonify({"errorCode": "VALIDATION_ERROR", "details": e.errors()}), 422, CORS_HEADERS except Exception as e: return jsonify({"errorCode": "BAD_REQUEST"}), 400, CORS_HEADERS try: add_email, status_code, headers = crud.add_email_to_user( user_id, parsed.email) if parsed.is_primary: # Neu parsed la nick email chinh # update email kratos identity = kratos.get_identity(user_id) traits = identity.traits traits["email"] = parsed.email res = kratos.update_identity( id=user_id, update_identity_body={ "schema_id": identity.schema_id, "traits": traits, "state": identity.state, }, ) return jsonify(add_email), status_code, headers except Exception as e: return jsonify({"error": str(e)}), 500 def user_filter_emails(): paging = EmailPage.from_request_queries() #paging là object EmailPage user_id = request.headers.get( "X-UserId") # X-Fission lay tren path if not user_id: return jsonify({"errorCode": "USER_ID_REQUIRED"}), 400, CORS_HEADERS conn = None try: conn = init_db_connection() with conn.cursor() as cursor: records = __filter_email(cursor, paging, user_id) # Xu ly gi day ? goi ham _filter_email return jsonify( records, ), 200, CORS_HEADERS except Exception as e: # current_app.logger.error(f"[filter_emails] DB Error: {e}") return jsonify({"errorCode": "DATABASE_ERROR"}), 500, CORS_HEADERS finally: if conn: conn.close() def __filter_email(cursor, paging: EmailPage, user_id: str): #Tất cả đều là đọc field trong object EmailFilter đặt trong field filter của EmailPage. # Truyen vao 3 tham so cursor, class EmailPage(Page+Filter), user_id kratos de check xem bao nhieu mail conditions = ["user_id = %(user_id)s"] # = SQL Where values = {"user_id": user_id} # user_id Kratos truyen vao if paging.filter.ids: #paging.filter là object EmailFilter bên trong paging conditions.append("id = ANY(%(ids)s)") # addpend them value vao conditions values["ids"] = paging.filter.ids if paging.filter.email: conditions.append("LOWER(email) LIKE %(email)s") values["email"] = f"%{paging.filter.email.lower()}%" if paging.filter.provider: conditions.append("LOWER(provider) LIKE %(provider)s") values["provider"] = f"%{paging.filter.provider.lower()}%" if paging.filter.created_from: conditions.append("created >= %(created_from)s") values["created_from"] = paging.filter.created_from if paging.filter.created_to: conditions.append("created <= %(created_to)s") values["created_to"] = paging.filter.created_to if paging.filter.modified_from: conditions.append("modified >= %(modified_from)s") values["modified_from"] = paging.filter.modified_from if paging.filter.modified_to: conditions.append("modified <= %(modified_to)s") values["modified_to"] = paging.filter.modified_to if paging.filter.keywords: conditions.append( "(LOWER(email) LIKE %(keywords)s OR LOWER(provider) LIKE %(keywords)s)") values["keywords"] = f"%{paging.filter.keywords.lower()}%" # conditions la 1 list cac doan dieu kien where where_clause = " AND ".join(conditions) if where_clause: # neu conditions co gia tri thi them WHERE neu khong thi khong them Where => Tránh thêm "WHERE " khi conditions rỗng where_clause = "WHERE " + where_clause order_clause = "" if paging.sortby: # neu paging.sortby ton tai direction = "ASC" if paging.asc else "DESC" order_clause = f"ORDER BY {paging.sortby} {direction}" sql = f""" SELECT *, COUNT(*) OVER() AS total FROM ailbl_user_email {where_clause} {order_clause} LIMIT %(limit)s OFFSET %(offset)s """ values["limit"] = paging.size values["offset"] = paging.page * paging.size cursor.execute(sql, values) return db_rows_to_array(cursor, cursor.fetchall())