gavindiaz/DinQuant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor code for improved readability and consistency

Browse Source 
- Cleaned up whitespace and formatting in various files including http.py, language.py, logger.py, safe_exec.py, and SQL migration scripts.
- Consolidated import statements and removed unnecessary blank lines.
- Updated logging configuration for better clarity.
- Enhanced the safe execution code with improved error handling and logging.
- Removed commented-out code and unnecessary variables in backfill_zero_trades.py and other scripts.
- Added a pyproject.toml for Ruff and Vulture configuration.
- Introduced requirements-dev.txt for development dependencies.
- Removed commented-out stock entries in init.sql for cleaner migration scripts.
This commit is contained in:
dienakdz
2026-04-09 14:30:51 +07:00
parent 103055b3df
commit 87f2845483
157 changed files with 19026 additions and 17773 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend_api_python/app/utils/auth.py
+72 -70
View File
@@ -4,44 +4,43 @@ Authentication Utilities
JWT token generation, verification, and middleware decorators.
Supports multi-user authentication with role-based access control.
"""
import jwt
import datetime
import os
from functools import wraps
from flask import request, jsonify, g
import jwt
from flask import g, jsonify, request
from app.config.settings import Config
from app.utils.logger import get_logger
logger = get_logger(__name__)
def generate_token(user_id: int, username: str, role: str = 'user', token_version: int = 1) -> str:
def generate_token(user_id: int, username: str, role: str = "user", token_version: int = 1) -> str:
"""
Generate JWT token with user information.
Args:
user_id: User ID
username: Username
role: User role (admin/manager/user/viewer)
token_version: Token version for single-client enforcement
Returns:
JWT token string
"""
try:
payload = {
'exp': datetime.datetime.utcnow() + datetime.timedelta(days=7),
'iat': datetime.datetime.utcnow(),
'sub': username,
'user_id': user_id,
'role': role,
'token_version': token_version, # For single client login control
"exp": datetime.datetime.utcnow() + datetime.timedelta(days=7),
"iat": datetime.datetime.utcnow(),
"sub": username,
"user_id": user_id,
"role": role,
"token_version": token_version, # For single client login control
}
return jwt.encode(
payload,
Config.SECRET_KEY,
algorithm='HS256'
)
return jwt.encode(payload, Config.SECRET_KEY, algorithm="HS256")
except Exception as e:
logger.error(f"Token generation failed: {e}")
return None
@@ -50,26 +49,26 @@ def generate_token(user_id: int, username: str, role: str = 'user', token_versio
def verify_token(token: str) -> dict:
"""
Verify JWT token and return payload.
Args:
token: JWT token string
Returns:
Token payload dict or None if invalid
"""
try:
payload = jwt.decode(token, Config.SECRET_KEY, algorithms=['HS256'])
payload = jwt.decode(token, Config.SECRET_KEY, algorithms=["HS256"])
# Verify token_version (single client login control)
user_id = payload.get('user_id')
token_version = payload.get('token_version')
user_id = payload.get("user_id")
token_version = payload.get("token_version")
if user_id and token_version is not None:
# Check if token_version in database matches
if not _verify_token_version(user_id, token_version):
logger.debug(f"Token version mismatch for user {user_id}: expected current, got {token_version}")
return None
return payload
except jwt.ExpiredSignatureError:
logger.debug("Token expired")
@@ -83,29 +82,27 @@ def _verify_token_version(user_id: int, token_version: int) -> bool:
"""
Verify that the token version matches the version stored in the database.
Used to implement single client login (kick out duplicate logins).
Args:
user_id: user ID
token_version: version number in Token
Returns:
True if version matches, False otherwise
"""
try:
from app.utils.db import get_db_connection
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"SELECT token_version FROM qd_users WHERE id = ?",
(user_id,)
)
cur.execute("SELECT token_version FROM qd_users WHERE id = ?", (user_id,))
row = cur.fetchone()
cur.close()
if not row:
return False
db_token_version = row.get('token_version') or 1
db_token_version = row.get("token_version") or 1
return int(token_version) == int(db_token_version)
except Exception as e:
logger.error(f"_verify_token_version failed: {e}")
@@ -115,45 +112,46 @@ def _verify_token_version(user_id: int, token_version: int) -> bool:
def get_current_user_id() -> int:
"""Get current user ID from flask.g context"""
return getattr(g, 'user_id', None)
return getattr(g, "user_id", None)
def get_current_user_role() -> str:
"""Get current user role from flask.g context"""
return getattr(g, 'user_role', 'user')
return getattr(g, "user_role", "user")
def login_required(f):
"""
Decorator that enforces Bearer token auth.
Sets g.user, g.user_id, g.user_role on successful auth.
"""
@wraps(f)
def decorated(*args, **kwargs):
token = None
# Read token from Authorization: Bearer <token>
auth_header = request.headers.get('Authorization')
auth_header = request.headers.get("Authorization")
if auth_header:
parts = auth_header.split()
if len(parts) == 2 and parts[0].lower() == 'bearer':
if len(parts) == 2 and parts[0].lower() == "bearer":
token = parts[1]
if not token:
return jsonify({'code': 401, 'msg': 'Token missing', 'data': None}), 401
return jsonify({"code": 401, "msg": "Token missing", "data": None}), 401
payload = verify_token(token)
if not payload:
return jsonify({'code': 401, 'msg': 'Token invalid or expired', 'data': None}), 401
return jsonify({"code": 401, "msg": "Token invalid or expired", "data": None}), 401
# Store user info in flask.g
g.user = payload.get('sub')
g.user_id = payload.get('user_id')
g.user_role = payload.get('role', 'user')
g.user = payload.get("sub")
g.user_id = payload.get("user_id")
g.user_role = payload.get("role", "user")
return f(*args, **kwargs)
return decorated
@@ -162,12 +160,14 @@ def admin_required(f):
Decorator that requires admin role.
Must be used after @login_required.
"""
@wraps(f)
def decorated(*args, **kwargs):
role = getattr(g, 'user_role', None)
if role != 'admin':
return jsonify({'code': 403, 'msg': 'Admin access required', 'data': None}), 403
role = getattr(g, "user_role", None)
if role != "admin":
return jsonify({"code": 403, "msg": "Admin access required", "data": None}), 403
return f(*args, **kwargs)
return decorated
@@ -176,12 +176,14 @@ def manager_required(f):
Decorator that requires manager or admin role.
Must be used after @login_required.
"""
@wraps(f)
def decorated(*args, **kwargs):
role = getattr(g, 'user_role', None)
if role not in ('admin', 'manager'):
return jsonify({'code': 403, 'msg': 'Manager access required', 'data': None}), 403
role = getattr(g, "user_role", None)
if role not in ("admin", "manager"):
return jsonify({"code": 403, "msg": "Manager access required", "data": None}), 403
return f(*args, **kwargs)
return decorated
@@ -189,38 +191,38 @@ def permission_required(permission: str):
"""
Decorator factory that checks for a specific permission.
Must be used after @login_required.
Usage:
@login_required
@permission_required('strategy')
def my_endpoint():
...
"""
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
role = getattr(g, 'user_role', 'user')
role = getattr(g, "user_role", "user")
# Import here to avoid circular import
from app.services.user_service import get_user_service
permissions = get_user_service().get_user_permissions(role)
if permission not in permissions:
return jsonify({
'code': 403,
'msg': f'Permission denied: {permission}',
'data': None
}), 403
return jsonify({"code": 403, "msg": f"Permission denied: {permission}", "data": None}), 403
return f(*args, **kwargs)
return decorated
return decorator
# Legacy compatibility: single-user mode fallback
def _is_single_user_mode() -> bool:
"""Check if system is in single-user (legacy) mode"""
return os.getenv('SINGLE_USER_MODE', 'false').lower() == 'true'
return os.getenv("SINGLE_USER_MODE", "false").lower() == "true"
def authenticate_legacy(username: str, password: str) -> dict:
@@ -230,9 +232,9 @@ def authenticate_legacy(username: str, password: str) -> dict:
"""
if username == Config.ADMIN_USER and password == Config.ADMIN_PASSWORD:
return {
'user_id': 1,
'username': username,
'role': 'admin',
'nickname': 'Admin',
"user_id": 1,
"username": username,
"role": "admin",
"nickname": "Admin",
}
return None