quangminhvs/py-ailbl-user-avatar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

User_Avatar_Done
Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 24s
Details

Browse Source
This commit is contained in:
QuangMinh_123
2025-12-02 04:38:34 +00:00
parent 02dc4943f6
commit 966e1c1aa8
13 changed files with 85 additions and 845 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.fission/dev-deployment.json
View File

@@ -1,7 +1,7 @@
{
"namespace": "default",
"secrets": {
"fission-ailbl-tag-env": {
"fission-ailbl-user-avatar-env": {
"literals": [
"PG_HOST=160.30.113.113",
"PG_PORT=45432",

16
.fission/local-deployment.json
View File

@@ -1,17 +1,13 @@
{
"namespace": "default",
"secrets": {
"fission-ailbl-tag-env": {
"fission-ailbl-user-avatar-env": {
"literals": [
"PG_HOST=160.30.113.113",
"PG_PORT=45432",
"PG_DB=postgres",
"PG_USER=postgres",
"PG_PASS=q2q32RQx9R9qVAp3vkVrrASnSUUhzKvC",
"MINIO_ENDPOINT=http://minio:9000",
"MINIO_ACCESS_KEY=minioadmin",
"MINIO_SECRET_KEY=minioadmin",
"MINIO_BUCKET=user-avatar"
"S3_BUCKET=ailbl",
"S3_ENDPOINT_URL=http://160.30.113.113:9000",
"S3_ACCESS_KEY_ID=quyen",
"S3_SECRET_ACCESS_KEY=12345678",
"S3_PREFIX=user/avatar"
]
}
}

2
.fission/staging-deployment.json
View File

@@ -1,7 +1,7 @@
{
"namespace": "default",
"secrets": {
"fission-ailbl-tag-env": {
"fission-ailbl-user-avatar-env": {
"literals": [
"PG_HOST=160.30.113.113",
"PG_PORT=45432",

2
.fission/test-deployment.json
View File

@@ -1,7 +1,7 @@
{
"namespace": "default",
"secrets": {
"fission-ailbl-tag-env": {
"fission-ailbl-user-avatar-env": {
"literals": [
"PG_HOST=160.30.113.113",
"PG_PORT=45432",

0
apps/__init__.py
View File

30
apps/ailbl-user_avatar-insert-update-delete-get.py
View File

@@ -4,14 +4,15 @@ from flask import jsonify, request
# from storage.minio_client import get_minio_client, check_existing_avatar_on_minio, upload_to_minio
ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
def main():
"""
```fission
{
"name": "avatar-admin-get-insert-delete-put",
"name": "avatar-users-get-insert-delete-put",
"http_triggers": {
"avatar-admin-get-insert-delete-put-http": {
"url": "/gh/users/avatars",
"avatar-users-get-insert-delete-put-http": {
"url": "/ailbl/users/avatars",
"methods": ["PUT", "POST", "DELETE", "GET"]
}
}
@@ -36,10 +37,12 @@ def main():
def make_insert_request():
try:
user_id = request.headers.get("X-User") # Lay user_id tu header X-User
file = request.files.get("avatar") #Lay file tu form-data voi key la 'avatar'
# Lay file tu form-data voi key la 'avatar'
file = request.files.get("avatar")
if not user_id or not file:
return jsonify({"error": "user_id or file is required"}), 400
if file.mimetype not in ALLOWED_IMAGE_TYPES: #Check mimetype(kieu du lieu cua file anh)
# Check mimetype(kieu du lieu cua file anh)
if file.mimetype not in ALLOWED_IMAGE_TYPES:
return jsonify(
{"error": "Invalid file type. Only JPG, PNG, GIF, WEBP are allowed."}
), 400
@@ -51,33 +54,36 @@ def make_insert_request():
def make_update_avatar_request():
try:
user_id = request.headers.get("X-User") # Lay user_id tu header X-User, neu co giao dien roi thi cookies se tu dong duoc gui len o trong header
file = request.files.get("avatar") #Lay file tu form-data voi key la 'avatar'
# Lay user_id tu header X-User, neu co giao dien roi thi cookies se tu dong duoc gui len o trong header
user_id = request.headers.get("X-User")
# Lay file tu form-data voi key la 'avatar'
file = request.files.get("avatar")
if not user_id or not file:
return jsonify({"error": "user_id or file is required"}), 400
if file.mimetype not in ALLOWED_IMAGE_TYPES: #Check mimetype(kieu du lieu cua file anh)
# Check mimetype(kieu du lieu cua file anh)
if file.mimetype not in ALLOWED_IMAGE_TYPES:
return jsonify(
{"error": "Invalid file type. Only JPG, PNG, GIF, WEBP are allowed."}
), 400
response, status = crud.update_or_create_avatar(user_id, file) # Call CRUD function to update avatar
response, status = crud.update_or_create_avatar(
user_id, file) # Call CRUD function to update avatar
return jsonify(response), status
except Exception as e:
return jsonify({"error": str(e)}), 500
def make_delete_avatar_request():
try:
user_id = request.headers.get("X-User") # Lay user_id tu header X-User
if not user_id:
return jsonify({"error": "user_id is required"}), 400
response, status = crud.delete_avatar(user_id) # Call CRUD function to delete avatar
# Call CRUD function to delete avatar
response, status = crud.delete_avatar(user_id)
return jsonify(response), status
except Exception as e:
return jsonify({"error": str(e)}), 500
def make_get_avatar_request():
try:
user_id = request.headers.get("X-User")

50
apps/crud.py
View File

@@ -1,23 +1,25 @@
import io
from flask import Response
from helpers import S3_BUCKET, get_secret, minio_client
from helpers import S3_BUCKET, get_secret, s3_client
from PIL import Image
def update_or_create_avatar(user_id: str, file): #Create&Update function to upload or update user avatar S3/Minio
# Create&Update function to upload or update user avatar S3/Minio
def update_or_create_avatar(user_id: str, file):
try:
file_data = file.read()
object_name = f"{get_secret('S3_PREFIX')}/{user_id}" # Bản chất là đường dẫn trong bucket + tên file = user_id
result = minio_client.put_object(
S3_BUCKET,
object_name,
io.BytesIO(file_data),
length=len(file_data),
content_type=file.content_type,
# Bản chất là đường dẫn trong bucket + tên file = user_id
object_name = f"{get_secret('S3_PREFIX')}/{user_id}"
result = s3_client.put_object(
Bucket=S3_BUCKET,
Key=object_name,
Body=io.BytesIO(file_data),
ContentLength=len(file_data),
ContentType=file.content_type,
)
return result.object_name, 200
return result, 200
except Exception as e:
return {"error": str(e)}, 500
@@ -25,28 +27,38 @@ def update_or_create_avatar(user_id: str, file): #Create&Update function to uplo
def get_avatar_url(user_id: str): # Read function to get user avatar from S3/Minio
try:
response = minio_client.get_object(
bucket_name=S3_BUCKET, object_name=f"{get_secret('S3_PREFIX')}/{user_id}"
response = s3_client.get_object(
Bucket=S3_BUCKET,
Key=f"{get_secret('S3_PREFIX')}/{user_id}"
)
image_data = response.read()
# image_data = response["body"].read(content_type)
image_data = response['Body'].read()
with Image.open(io.BytesIO(image_data)) as img:
fmt = img.format.lower() # ví dụ: 'jpeg', 'png', 'webp'
content_type = f"image/{'jpeg' if fmt == 'jpg' else fmt}"
# return Response(
# io.BytesIO(image_data),
# content_type=content_type,
# direct_passthrough=True,
# )
return Response(
io.BytesIO(image_data),
image_data,
content_type=content_type,
direct_passthrough=True,
)
direct_passthrough=True
), 200
except Exception as e:
return {"error": str(e)}, 500
def delete_avatar(user_id: str) -> dict: #Delete Function to delete user avatar from S3/Minio
# Delete Function to delete user avatar from S3/Minio
def delete_avatar(user_id: str) -> dict:
try:
result = minio_client.remove_object(
S3_BUCKET, f"{get_secret('S3_PREFIX')}/{user_id}"
result = s3_client.delete_object(
Bucket=S3_BUCKET,
Key=f"{get_secret('S3_PREFIX')}/{user_id}"
)
return result, 200
except Exception as e:

126
apps/helpers.py
View File

@@ -1,133 +1,35 @@
import datetime
import logging
import socket
import os
import psycopg2
from flask import current_app
from psycopg2.extras import LoggingConnection
from minio import Minio
CORS_HEADERS = {
"Content-Type": "application/json",
}
import boto3
SECRET_NAME = "fission-ailbl-user-avatar-env"
CONFIG_NAME = "fission-eom-notification-config"
K8S_NAMESPACE = "default"
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
def init_db_connection():
db_host = get_secret("PG_HOST", "127.0.0.1")
db_port = int(get_secret("PG_PORT", 5432))
if not check_port_open(ip=db_host, port=db_port):
raise Exception(f"Establishing A Database Connection. `{db_host}:{db_port}`")
# options = get_secret("PG_DBSCHEMA")
# if options:
# options = f"-c search_path={options}" # if specific db schema
conn = psycopg2.connect(
database=get_secret("PG_DB", "postgres"),
user=get_secret("PG_USER", "postgres"),
password=get_secret("PG_PASS", "secret"),
host=get_secret("PG_HOST", "127.0.0.1"),
port=int(get_secret("PG_PORT", 5432)),
# options=options,
# cursor_factory=NamedTupleCursor,
connection_factory=LoggingConnection,
)
conn.initialize(logger)
return conn
def db_row_to_dict(cursor, row):
record = {}
for i, column in enumerate(cursor.description):
data = row[i]
if isinstance(data, datetime.datetime):
data = data.isoformat()
record[column.name] = data
return record
def db_rows_to_array(cursor, rows):
return [db_row_to_dict(cursor, row) for row in rows]
def get_current_namespace() -> str: # Lấy config từ env fission kubernetes => Vì có nhiều môi trường khác nhau
def get_current_namespace() -> str:
try:
with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace", "r") as f:
namespace = f.read()
except Exception as err:
current_app.logger.error(err)
except:
namespace = K8S_NAMESPACE
return str(namespace)
def get_secret(key: str, default=None):
def get_secret(key: str, default=None) -> str:
namespace = get_current_namespace()
path = f"/secrets/{namespace}/{SECRET_NAME}/{key}"
try:
with open(path, "r") as f:
return f.read()
except Exception as err:
current_app.logger.error(path, err)
except:
return default
S3_BUCKET = get_secret("S3_BUCKET")
def get_minio_client():
client = Minio(
os.getenv("MINIO_ENDPOINT", "localhost:9000"),
access_key=os.getenv("MINIO_ACCESS_KEY", "minioadmin"),
secret_key=os.getenv("MINIO_SECRET_KEY", "minioadmin"),
secure=False
S3_PREFIX = get_secret("S3_PREFIX")
s3_client = boto3.client(
"s3",
endpoint_url=get_secret("S3_ENDPOINT_URL"),
aws_access_key_id=get_secret("S3_ACCESS_KEY_ID"),
aws_secret_access_key=get_secret("S3_SECRET_ACCESS_KEY"),
config=boto3.session.Config(signature_version="s3v4"),
)
return client
def get_config(key: str, default=None):
namespace = get_current_namespace()
path = f"/configs/{namespace}/{CONFIG_NAME}/{key}"
try:
with open(path, "r") as f:
return f.read()
except Exception as err:
current_app.logger.error(path, err)
return default
def str_to_bool(input: str | None) -> bool:
input = input or ""
# Dictionary to map string values to boolean
BOOL_MAP = {"true": True, "false": False}
return BOOL_MAP.get(input.strip().lower(), None)
def check_port_open(ip: str, port: int, timeout: int = 30):
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(timeout)
result = s.connect_ex((ip, port))
return result == 0
except Exception as err:
current_app.logger.err(f"Check port open error: {err}")
return False
# # Get DB connection
# def get_db_connection(cursor_factory=None): # Hàm truy cập đến database, Thông tin database sẽ nằm ở trong này
# try:
# conn = psycopg2.connect(
# host=os.getenv("DB_HOST"),
# database=os.getenv("DB_NAME"), #biến env(môi trường)
# user=os.getenv("DB_USERNAME"),
# password=os.getenv("DB_PASSWORD")
# )
# print("✅ Connected to DB successfully")
# return conn
# except Exception as e:
# print("❌ Database connection failed:", e)
# raise

3
apps/requirements.txt
View File

@@ -1,3 +1,6 @@
Flask==3.1.0
psycopg2-binary==2.9.10
pydantic==2.11.3
minio==7.2.5
Pillow==10.4.0
boto3==1.35.70

279
apps/tag_filter_insert.py
View File

@@ -1,279 +0,0 @@
import dataclasses
import enum
import json
import typing
import uuid
from flask import current_app, jsonify, request
from helpers import (
CORS_HEADERS,
db_row_to_dict,
db_rows_to_array,
init_db_connection,
str_to_bool,
)
from pydantic import ValidationError
from schemas import TagRequest
def main():
"""
Filter or Insert Notification by id
```fission
{
"name": "ailbl-tag-admin-filter-or-insert",
"http_triggers": {
"ailbl-tag-admin-filter-or-insert-http": {
"url": "/ailbl/admin/tags",
"methods": ["GET", "POST"]
}
}
}
```
"""
try:
if request.method == "GET":
return make_filter_request()
elif request.method == "POST":
return make_insert_request()
else:
return {"error": "Method not allow"}, 405, CORS_HEADERS
except KeyError as _err:
sortby_variables = [e.name for e in TagSortField]
type_variables = [e.name for e in KindType]
return (
{
"error": f"SortBy should be in {sortby_variables} and KindType should be in {type_variables}"
# "error": f"SortBy should be in {sortby_variables} and NotiType should be in"
},
400,
CORS_HEADERS,
)
except Exception as err:
print(f"ErrorType={type(err)}")
return {"error": str(err)}, 500, CORS_HEADERS
def make_insert_request():
conn = None
try:
try:
request_data = request.get_json()
data = TagRequest(**request_data)
except ValidationError as e:
return jsonify({"error": "Validation failed", "details": e.errors()}), 400
conn = init_db_connection()
with conn.cursor() as cursor:
tag = __insert_tag(cursor, data)
conn.commit()
return jsonify(tag)
except Exception as ex:
return jsonify({"error": str(ex)}), 500
finally:
if conn is not None:
conn.close()
current_app.logger.info("Close DB connection")
def __insert_tag(cursor, data: TagRequest):
sql = """
INSERT INTO ailbl_tag (id, tag, kind, ref, primary_color, secondary_color, created, modified)
VALUES (%(id)s, %(tag)s, %(kind)s, %(ref)s, %(primary_color)s, %(secondary_color)s, now(), now())
RETURNING *
"""
cursor.execute(
sql,
{
"id": str(uuid.uuid4()),
"tag": data.tag,
"kind": data.kind.value,
"ref": data.ref,
"primary_color": data.primary_color,
"secondary_color": data.secondary_color,
},
)
row = cursor.fetchone()
if row:
return db_row_to_dict(cursor, row)
else:
raise Exception("Insert tag failed")
def make_filter_request():
paging = TagPage.from_request_queries()
conn = None
try:
conn = init_db_connection()
with conn.cursor() as cursor:
records = __filter_tag(cursor, paging)
return jsonify(records)
finally:
if conn is not None:
conn.close()
current_app.logger.info("Close DB connection")
def __filter_tag(cursor, paging: "TagPage"):
conditions = []
values = {}
if paging.filter.ids:
conditions.append("id = ANY(%(ids)s)")
values["ids"] = paging.filter.ids
if paging.filter.keyword:
conditions.append("LOWER(tag) LIKE %(keyword)s")
values["keyword"] = f"%{paging.filter.keyword.lower()}%"
if paging.filter.kind:
conditions.append("kind = %(kind)s::smallint")
values["kind"] = int(paging.filter.kind)
if paging.filter.ref:
conditions.append("LOWER(ref) LIKE %(ref)s")
values["ref"] = f"%{paging.filter.ref.lower()}%"
if paging.filter.primary_color:
conditions.append("primary_color = %(primary_color)s")
values["primary_color"] = paging.filter.primary_color
if paging.filter.secondary_color:
conditions.append("secondary_color = %(secondary_color)s")
values["secondary_color"] = paging.filter.secondary_color
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
where_clause = " AND ".join(conditions)
if where_clause:
where_clause = "WHERE " + where_clause
order_clause = ""
if paging.sortby:
direction = "ASC" if paging.asc else "DESC"
order_clause = f" ORDER BY {paging.sortby.value} {direction} "
sql = f"""
SELECT
t.*,
(
SELECT COUNT(*)
FROM ailbl_tag_ref r
WHERE r.tag_id = t.id
) AS ref_count,
count(*) OVER() AS total
FROM ailbl_tag t
{where_clause}
{order_clause}
LIMIT %(limit)s OFFSET %(offset)s
"""
values["limit"] = paging.size
values["offset"] = paging.page * paging.size
cursor.execute(sql, values)
rows = cursor.fetchall()
return db_rows_to_array(cursor, rows)
@dataclasses.dataclass
class Page:
page: typing.Optional[int] = None
size: typing.Optional[int] = None
asc: typing.Optional[bool] = None
@classmethod
def from_request_queries(cls) -> "Page":
paging = Page()
paging.page = int(request.args.get("page", 0))
paging.size = int(request.args.get("size", 8))
paging.asc = request.args.get("asc", type=str_to_bool)
return paging
class TagSortField(str, enum.Enum):
CREATED = "created"
KIND = "kind"
MODIFIED = "modified"
class KindType(str, enum.Enum):
"""Notification Types"""
PROJECT_GROUP = "1"
PROJECT_DATA = "2"
PROJECT_MEMBER = "3"
PROJECT_DISCUSSTION_TOPIC = "4"
PROJECT = "5"
TICKET = "6"
@dataclasses.dataclass
class TagFilter:
ids: typing.Optional[typing.List[str]] = None
tag: typing.Optional[str] = None
keyword: typing.Optional[str] = None
kind: typing.Optional[typing.List[KindType]] = None
ref: typing.Optional[str] = None
primary_color: typing.Optional[str] = None
secondary_color: 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
@classmethod
def from_request_queries(cls) -> "TagFilter":
filter = TagFilter()
filter.ids = request.args.getlist("filter[ids]")
filter.keyword = request.args.get("filter[keyword]")
kind_str = request.args.get("filter[kind]")
if kind_str:
try:
kind_enum = KindType[kind_str] # enum theo .name
filter.kind = kind_enum.value # .value là "1", "2", ...
except KeyError:
raise ValueError(
f"KindType should be one of: {[e.name for e in KindType]}"
)
filter.ref = request.args.get("filter[ref]")
filter.primary_color = request.args.get("filter[primary_color]")
filter.secondary_color = request.args.get("filter[secondary_color]")
filter.created_to = request.args.get("filter[created_to]")
filter.created_from = request.args.get("filter[created_from]")
filter.modified_from = request.args.get("filter[modified_from]")
filter.modified_to = request.args.get("filter[modified_to]")
return filter
@dataclasses.dataclass
class TagPage(Page):
sortby: typing.Optional[TagSortField] = None
filter: typing.Optional[TagFilter] = dataclasses.field(
default_factory=TagFilter.from_request_queries
)
@classmethod
def from_request_queries(cls) -> "TagPage":
paging = super(TagPage, cls).from_request_queries()
paging = TagPage(**dataclasses.asdict(paging))
paging.sortby = (
TagSortField[request.args.get("sortby")]
if request.args.get("sortby")
else None
)
return paging

144
apps/tag_update_delete.py
View File

@@ -1,144 +0,0 @@
import json
from flask import current_app, jsonify, request
from helpers import CORS_HEADERS, db_row_to_dict, init_db_connection
from pydantic import ValidationError
from schemas import TagRequestUpdate
def main():
"""
Update or Delete Notification by id
```fission
{
"name": "ailbl-tag-admin-update-or-delete",
"http_triggers": {
"ailbl-tag-admin-update-or-delete-http": {
"url": "/ailbl/admin/tags/{TagId}",
"methods": ["PUT", "DELETE"]
}
}
}
```
"""
try:
if request.method == "PUT":
return make_update_request()
elif request.method == "DELETE":
return make_delete_request()
else:
return {"error": "Method not allow"}, 405, CORS_HEADERS
except Exception as err:
return {"error": str(err)}, 500, CORS_HEADERS
def make_delete_request():
tag_id = request.headers.get("X-Fission-Params-TagId")
if not tag_id:
return jsonify({"errorCode": "TAG_ID_REQUIRED"}), 400
conn = None
try:
conn = init_db_connection()
with conn.cursor() as cursor:
result = __delete_tag(cursor, id=tag_id)
if result == "TAG_NOT_FOUND":
return jsonify({"errorCode": "TAG_NOT_FOUND"}), 404
if result == "TAG_HAS_REF":
return jsonify({"errorCode": "TAG_HAS_REF"}), 400
conn.commit()
return jsonify(result), 200
except Exception as ex:
return jsonify({"error": str(ex)}), 500
finally:
if conn is not None:
conn.close()
current_app.logger.info("Close DB connection")
def make_update_request():
tag_id = request.headers.get("X-Fission-Params-TagId")
if not tag_id:
return jsonify({"errorCode": "TAG_ID_REQUIRED"}), 400
conn = None
try:
try:
request_data = request.get_json()
data = TagRequestUpdate(**request_data)
except ValidationError as e:
return jsonify({"error": "Validation failed", "details": e.errors()}), 400
with init_db_connection() as conn:
with conn.cursor() as cursor:
existed = __get_tag(cursor, id=tag_id)
if not existed:
return jsonify({"errorCode": "TAG_NOT_FOUND"}), 404
record = __update_tag(cursor, tag_id, data)
conn.commit()
return jsonify(record)
except Exception as ex:
return jsonify({"error": str(ex)}), 500
finally:
if conn is not None:
conn.close()
current_app.logger.info("Close DB connection")
def __update_tag(cursor, tag_id: str, data: TagRequestUpdate):
set_fields = ["modified = CURRENT_TIMESTAMP"]
values = {"id": tag_id}
if data.tag:
set_fields.append("tag = %(tag)s")
values["tag"] = data.tag
if data.kind:
set_fields.append("kind = %(kind)s::smallint")
values["kind"] = data.kind.value
if data.ref is not None:
set_fields.append("ref = %(ref)s")
values["ref"] = data.ref
if data.primary_color is not None:
set_fields.append("primary_color = %(primary_color)s")
values["primary_color"] = data.primary_color
if data.secondary_color is not None:
set_fields.append("secondary_color = %(secondary_color)s")
values["secondary_color"] = data.secondary_color
if len(set_fields) == 1:
raise Exception("Nothing to update")
sql = f"""
UPDATE ailbl_tag
SET {", ".join(set_fields)}
WHERE id = %(id)s
RETURNING *
"""
cursor.execute(sql, values)
if row := cursor.fetchone():
return db_row_to_dict(cursor, row)
else:
raise Exception(f"Tag with id={tag_id} not found or unchanged")
def __delete_tag(cursor, id: str):
cursor.execute("SELECT 1 FROM ailbl_tag WHERE id = %(id)s", {"id": id})
if not cursor.fetchone():
return "TAG_NOT_FOUND"
cursor.execute(
"SELECT 1 FROM ailbl_tag_ref WHERE tag_id = %(id)s LIMIT 1", {"id": id}
)
if cursor.fetchone():
return "TAG_HAS_REF"
cursor.execute("DELETE FROM ailbl_tag WHERE id = %(id)s RETURNING *", {"id": id})
row = cursor.fetchone()
return db_row_to_dict(cursor, row)
def __get_tag(cursor, id: str):
sql = """ SELECT * FROM ailbl_tag WHERE id=%(id)s """
cursor.execute(sql, {"id": id})
row = cursor.fetchone()
return db_row_to_dict(cursor, row) if row else None

256
apps/tagref_filter_filter_insert.py
View File

@@ -1,256 +0,0 @@
import dataclasses
import enum
# import json
import typing
import uuid
from flask import current_app, jsonify, request
from helpers import (
CORS_HEADERS,
db_row_to_dict,
db_rows_to_array,
init_db_connection,
str_to_bool,
)
from pydantic import ValidationError
from schemas import TagRefRequest
def main():
"""
```fission
{
"name": "ailbl-tag-ref-admin-filter-or-insert",
"http_triggers": {
"ailbl-tag-ref-admin-filter-or-insert-http": {
"url": "/ailbl/admin/tags/{TagId}/refs",
"methods": ["GET", "POST"]
}
}
}
```
"""
try:
if request.method == "GET":
return make_filter_request()
elif request.method == "POST":
return make_insert_request()
else:
return {"error": "Method not allow"}, 405, CORS_HEADERS
except KeyError as _err:
sortby_variables = [e.name for e in TagSortField]
return (
{"error": f"SortBy should be in {sortby_variables}"},
400,
CORS_HEADERS,
)
except Exception as err:
print(f"ErrorType={type(err)}")
return {"error": str(err)}, 500, CORS_HEADERS
def make_insert_request():
conn = None
try:
tag_id = request.headers.get("X-Fission-Params-TagId")
if not tag_id:
return jsonify({"errorCode": "TAG_ID_REQUIRED"}), 400
try:
request_data = request.get_json()
data = TagRefRequest(**request_data)
except ValidationError as e:
return jsonify({"error": "Validation failed", "details": e.errors()}), 400
conn = init_db_connection()
with conn.cursor() as cursor:
tag = __insert_tag_ref(cursor, tag_id, data)
conn.commit()
return jsonify(tag)
except Exception as ex:
return jsonify({"error": str(ex)}), 500
finally:
if conn is not None:
conn.close()
current_app.logger.info("Close DB connection")
def __insert_tag_ref(cursor, tag_id: str, data: TagRefRequest):
sql = """
INSERT INTO ailbl_tag_ref (id, tag_id, ref, sub_ref)
VALUES (%(id)s, %(tag_id)s, %(ref)s, %(sub_ref)s)
RETURNING *
"""
cursor.execute(
sql,
{
"id": str(uuid.uuid4()),
"tag_id": tag_id,
"ref": data.ref,
"sub_ref": data.sub_ref,
},
)
row = cursor.fetchone()
if row:
return db_row_to_dict(cursor, row)
else:
raise Exception("Insert tag_ref failed")
def make_filter_request():
paging = TagRefPage.from_request_queries()
tag_id = request.headers.get("X-Fission-Params-TagId")
if not tag_id:
return jsonify({"errorCode": "TAG_ID_REQUIRED"}), 400
conn = None
try:
conn = init_db_connection()
with conn.cursor() as cursor:
records = __filter_tag_ref(cursor, tag_id, paging)
return jsonify(records)
finally:
if conn is not None:
conn.close()
current_app.logger.info("Close DB connection")
def __filter_tag_ref(cursor, tag_id: str, paging: "TagRefPage"):
conditions = ["tag_id = %(tag_id)s"]
values = {"tag_id": tag_id}
if paging.filter.ids:
conditions.append("id = ANY(%(ids)s)")
values["ids"] = paging.filter.ids
# if paging.filter.tag_ids:
# conditions.append("tag_id = ANY(%(tag_ids)s)")
# values["tag_ids"] = paging.filter.tag_ids
if paging.filter.ref:
conditions.append("LOWER(ref) LIKE %(keyword)s")
values["ref"] = f"%{paging.filter.ref.lower()}%"
where_clause = " AND ".join(conditions)
order_clause = ""
if paging.sortby:
direction = "ASC" if paging.asc else "DESC"
order_clause = f" ORDER BY {paging.sortby.value} {direction}"
sql = f"""
SELECT *, count(*) OVER() AS total
FROM ailbl_tag_ref
WHERE {where_clause}
{order_clause}
LIMIT %(limit)s OFFSET %(offset)s
"""
values["limit"] = paging.size
values["offset"] = paging.page * paging.size
cursor.execute(sql, values)
rows = cursor.fetchall()
return db_rows_to_array(cursor, rows)
@dataclasses.dataclass
class Page:
page: typing.Optional[int] = None
size: typing.Optional[int] = None
asc: typing.Optional[bool] = None
@classmethod
def from_request_queries(cls) -> "Page":
paging = Page()
paging.page = int(request.args.get("page", 0))
paging.size = int(request.args.get("size", 8))
paging.asc = request.args.get("asc", type=str_to_bool)
return paging
class TagSortField(str, enum.Enum):
TAG = "tag_id"
REF = "ref"
@dataclasses.dataclass
class TagFilter:
ids: typing.Optional[typing.List[str]] = None
tag_ids: typing.Optional[typing.List[str]] = None
ref: typing.Optional[str] = None
@classmethod
def from_request_queries(cls) -> "TagFilter":
filter = TagFilter()
filter.ids = request.args.getlist("filter[ids]")
filter.tag_ids = request.args.get("filter[tag_ids]")
filter.ref = request.args.get("filter[ref]")
return filter
@dataclasses.dataclass
class TagRefPage(Page):
sortby: typing.Optional[TagSortField] = None
filter: typing.Optional[TagFilter] = dataclasses.field(
default_factory=TagFilter.from_request_queries
)
@classmethod
def from_request_queries(cls) -> "TagRefPage":
paging = super(TagRefPage, cls).from_request_queries()
paging = TagRefPage(**dataclasses.asdict(paging))
paging.sortby = (
TagSortField[request.args.get("sortby")]
if request.args.get("sortby")
else None
)
return paging
def make_delete_request():
"""
```fission
{
"name": "ailbl-tag-ref-admin-delete",
"http_triggers": {
"ailbl-tag-ref-admin-delete-http": {
"url": "/ailbl/admin/tags/{TagId}/refs/{TagRefId}",
"methods": ["DELETE"]
}
}
}
```
"""
if request.method != "DELETE":
return {"error": "Method not allow"}, 405, CORS_HEADERS
tag_id = request.headers.get("X-Fission-Params-TagId")
tag_ref_id = request.headers.get("X-Fission-Params-TagRefId")
if not tag_id:
return jsonify({"errorCode": "TAG_ID_REQUIRED"}), 400
if not tag_ref_id:
return jsonify({"errorCode": "TAG_REF_ID_REQUIRED"}), 400
conn = None
try:
with init_db_connection() as conn:
with conn.cursor() as cursor:
record = __delete_tag_ref(cursor, tag_ref_id=tag_ref_id, tag_id=tag_id)
if not record:
return jsonify({"errorCode": "TAG_REF_NOT_FOUND"}), 404
return jsonify(record)
conn.commit()
except Exception as ex:
return jsonify({"error": str(ex)}), 500
finally:
if conn is not None:
conn.close()
current_app.logger.info("Close DB connection")
def __delete_tag_ref(cursor, tag_ref_id: str, tag_id: str):
sql = """
DELETE FROM ailbl_tag_ref
WHERE id = %(tag_ref_id)s AND tag_id = %(tag_id)s
RETURNING *
"""
cursor.execute(sql, {"tag_ref_id": tag_ref_id, "tag_id": tag_id})
row = cursor.fetchone()
return db_row_to_dict(cursor, row) if row else None