sipro blueprint

routes/sipro.py

@@ -0,0 +1,53 @@
+from enum import StrEnum
+
+from flask import Blueprint, request
+from sqlalchemy import func
+
+import auxiliary
+import database
+
+sipro_blueprint = Blueprint('sipro', __name__)
+
+
+class ExpandParam(StrEnum):
+    USER = 'user'
+
+
+@sipro_blueprint.before_request
+def auth():
+    API_KEY = '5D809ED08080B5F204443B31374BD6A5'
+    if request.headers.get('Authorization') != API_KEY:
+        return {'error': 'Unauthorized'}, 401
+
+
+@sipro_blueprint.post('/assemblyInfo')
+def assembly_info():
+    params: dict = request.json
+    expand_param = params.get('expand')
+    order_ids = params.get('orderIds')
+    expand_list = expand_param if expand_param else []
+    datetime_format = 'YYYY-MM-DD HH24:MI:SS'
+    entity_list = [
+        database.Assembly.id.label('id'),
+        database.Assembly.order_id.label('order_id'),
+        func.to_char(database.Assembly.created_at, datetime_format).label('created_at'),
+        func.to_char(database.Assembly.ended_at, datetime_format).label('ended_at'),
+        database.Assembly.is_active.label('is_active'),
+        database.Assembly.state.label('state')
+    ]
+    query = database.Assembly.query
+    for expand in expand_list:
+        match expand:
+            case ExpandParam.USER:
+                query = query.join(database.User)
+                entity_list.extend([
+                    database.User.id.label('user.id'),
+                    database.User.login.label('user.login'),
+                ])
+
+    if order_ids:
+        query = query.filter(database.Assembly.order_id.in_(order_ids))
+    query = query.with_entities(*entity_list)
+    result = query.all()
+    json_result = [auxiliary.to_nested_dict(row) for row in result]
+    return json_result