icecheng/freeleaps-service-hub
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: fix the invalid id format issues

Browse Source
This commit is contained in:
haolou 2025-10-23 15:53:04 +08:00
parent 1823a33d45
commit ba63e1b7a8
3 changed files with 36 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
apps/authentication/backend/infra/permission/permission_handler.py
View File

@ -132,9 +132,9 @@ class PermissionHandler:
"""Query permissions with pagination and fuzzy search"""
query = {}
if permission_key:
query[str(PermissionDoc.permission_key)] = {"$regex": permission_key, "$options": "i"}
query["permission_key"] = {"$regex": permission_key, "$options": "i"}
if permission_name:
query[str(PermissionDoc.permission_name)] = {"$regex": permission_name, "$options": "i"}
query["permission_name"] = {"$regex": permission_name, "$options": "i"}
cursor = PermissionDoc.find(query)
total = await cursor.count()
docs = await cursor.skip(skip).limit(limit).to_list()
@ -150,13 +150,14 @@ class PermissionHandler:
query = {}
if permission_id:
try:
query[str(PermissionDoc.id)] = permission_id
ObjectId(permission_id) # Validate ObjectId format
query["_id"] = permission_id # Use MongoDB's _id field directly
except Exception:
raise RequestValidationError("Invalid permission_id format. Must be a valid ObjectId.")
if permission_key:
query[str(PermissionDoc.permission_key)] = {"$regex": permission_key, "$options": "i"}
query["permission_key"] = {"$regex": permission_key, "$options": "i"}
if permission_name:
query[str(PermissionDoc.permission_name)] = {"$regex": permission_name, "$options": "i"}
query["permission_name"] = {"$regex": permission_name, "$options": "i"}
cursor = PermissionDoc.find(query)
total = await cursor.count()
docs = await cursor.to_list()

11
apps/authentication/backend/infra/permission/role_handler.py
View File

@ -126,9 +126,9 @@ class RoleHandler:
"""Query roles with pagination and fuzzy search by role_key and role_name"""
query = {}
if role_key:
query[str(RoleDoc.role_key)] = {"$regex": role_key, "$options": "i"}
query["role_key"] = {"$regex": role_key, "$options": "i"}
if role_name:
query[str(RoleDoc.role_name)] = {"$regex": role_name, "$options": "i"}
query["role_name"] = {"$regex": role_name, "$options": "i"}
cursor = RoleDoc.find(query)
total = await cursor.count()
docs = await cursor.skip(skip).limit(limit).to_list()
@ -144,13 +144,14 @@ class RoleHandler:
query = {}
if role_id:
try:
query[str(RoleDoc.id)] = role_id
ObjectId(role_id) # Validate ObjectId format
query["_id"] = role_id # Use MongoDB's _id field directly
except Exception:
raise RequestValidationError("Invalid role_id format. Must be a valid ObjectId.")
if role_key:
query[str(RoleDoc.role_key)] = {"$regex": role_key, "$options": "i"}
query["role_key"] = {"$regex": role_key, "$options": "i"}
if role_name:
query[str(RoleDoc.role_name)] = {"$regex": role_name, "$options": "i"}
query["role_name"] = {"$regex": role_name, "$options": "i"}
cursor = RoleDoc.find(query)
total = await cursor.count()
docs = await cursor.to_list()

38
apps/authentication/backend/models/base_doc.py
View File

@ -6,6 +6,7 @@ from datetime import datetime, timezone
from typing import Optional, List, Dict, Any, Type, Union
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
from pydantic import BaseModel
from pydantic._internal._model_construction import ModelMetaclass
from common.config.app_settings import app_settings
@ -14,27 +15,27 @@ class QueryExpression:
def __init__(self, field_name: str):
self.field_name = field_name
def __eq__(self, other) -> dict:
def __eq__(self, other: Any) -> Dict[str, Any]:
"""Handle field == value comparisons"""
return {self.field_name: other}
def __ne__(self, other) -> dict:
def __ne__(self, other: Any) -> Dict[str, Any]:
"""Handle field != value comparisons"""
return {self.field_name: {"$ne": other}}
def __gt__(self, other) -> dict:
def __gt__(self, other: Any) -> Dict[str, Any]:
"""Handle field > value comparisons"""
return {self.field_name: {"$gt": other}}
def __lt__(self, other) -> dict:
def __lt__(self, other: Any) -> Dict[str, Any]:
"""Handle field < value comparisons"""
return {self.field_name: {"$lt": other}}
def __ge__(self, other) -> dict:
def __ge__(self, other: Any) -> Dict[str, Any]:
"""Handle field >= value comparisons"""
return {self.field_name: {"$gte": other}}
def __le__(self, other) -> dict:
def __le__(self, other: Any) -> Dict[str, Any]:
"""Handle field <= value comparisons"""
return {self.field_name: {"$lte": other}}
@ -80,13 +81,13 @@ import contextvars
_tenant_db_context: contextvars.ContextVar[Optional[AsyncIOMotorDatabase]] = contextvars.ContextVar('tenant_db', default=None)
class QueryModelMeta(type(BaseModel)):
class QueryModelMeta(ModelMetaclass):
"""Metaclass: automatically create FieldDescriptor for model fields"""
def __new__(cls, name: str, bases: tuple, namespace: dict):
# Get model field annotations (like name: str -> "name" and str)
annotations = namespace.get("__annotations__", {})
# Create the class first
# Create the class first using Pydantic's metaclass
new_class = super().__new__(cls, name, bases, namespace)
# After Pydantic processes the fields, add the descriptors as class attributes
@ -97,6 +98,13 @@ class QueryModelMeta(type(BaseModel)):
return new_class
def __getattr__(cls, name: str):
"""Handle field access like Doc.field_name for query building"""
# Check if this is a field that exists in the model
if hasattr(cls, 'model_fields') and name in cls.model_fields:
return QueryExpression(name)
raise AttributeError(f"'{cls.__name__}' object has no attribute '{name}'")
class BaseDoc(BaseModel, metaclass=QueryModelMeta):
"""
Base document class that provides Beanie-like interface using direct MongoDB operations.
@ -118,8 +126,10 @@ class BaseDoc(BaseModel, metaclass=QueryModelMeta):
return filtered_result
@classmethod
def field(cls, field_name: str) -> QueryExpression:
"""Get a field expression for query building"""
return QueryExpression(field_name)
@classmethod
async def _get_database(cls) -> AsyncIOMotorDatabase:
@ -128,14 +138,14 @@ class BaseDoc(BaseModel, metaclass=QueryModelMeta):
tenant_db = _tenant_db_context.get()
if tenant_db is not None:
return tenant_db
# Fallback to global database connection
global _db, _client
if _db is None:
_client = AsyncIOMotorClient(app_settings.MONGODB_URI)
_db = _client[app_settings.MONGODB_NAME]
return _db
@classmethod
def set_tenant_database(cls, db: AsyncIOMotorDatabase):
"""Set the tenant database for this context"""
@ -344,8 +354,8 @@ class QueryBuilder:
def __init__(self, model_class: Type[BaseDoc], conditions: tuple):
self.model_class = model_class
self.conditions = conditions
self._limit_value = None
self._skip_value = None
self._limit_value: Optional[int] = None
self._skip_value: Optional[int] = None
def limit(self, n: int) -> 'QueryBuilder':
"""Limit number of results"""