Performance Optimierungen: Rate Limit, created_at Index, Keyset-Pagination

1. Rate Limit erhoeht: 60 -> 300 Requests/Minute (fuer 100+ User)
2. Migration 0099: created_at DESC Index auf allen Tabellen (Order by Performance)
3. Keyset-Pagination: optionaler cursor Parameter fuer contacts API
   - cursor=UUID nutzt WHERE id > cursor statt OFFSET
   - Backward compatible: ohne cursor wird page/page_size genutzt
   - next_cursor in Response fuer naechste Seite

Tests: 43/43 bestanden
This commit is contained in:
Agent Zero
2026-08-03 19:18:09 +02:00
parent 67c05f39f1
commit 5d35f0064e
4 changed files with 178 additions and 4 deletions
+1 -1
View File
@@ -73,7 +73,7 @@ class Settings(BaseSettings):
rate_limit_reset_window: int = 3600 # 1 hour
rate_limit_reset_confirm_max: int = 5
rate_limit_reset_confirm_window: int = 3600 # 1 hour
rate_limit_general_max: int = 60
rate_limit_general_max: int = 300
rate_limit_general_window: int = 60 # 1 min
@property
+6 -1
View File
@@ -65,10 +65,14 @@ async def list_contacts(
folder_id: str | None = Query(None),
sort_by: str = Query("displayname"),
sort_order: str = Query("asc", pattern="^(asc|desc)$"),
cursor: str | None = Query(None, description="Keyset pagination cursor (contact UUID)"),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("contacts:read")),
):
"""List contacts with pagination, FTS search, type/folder filter, sorting."""
"""List contacts with pagination, FTS search, type/folder filter, sorting.
Supports keyset pagination via ``cursor`` parameter for large datasets.
"""
tenant_id = uuid.UUID(current_user["tenant_id"])
user_id = uuid.UUID(current_user["user_id"])
is_admin = current_user.get("is_system_admin", False)
@@ -80,6 +84,7 @@ async def list_contacts(
resolved_perms=current_user,
user_id=user_id,
is_system_admin=is_admin,
cursor=cursor,
)
+22 -2
View File
@@ -147,10 +147,16 @@ async def list_contacts(
resolved_perms: dict | None = None,
user_id: uuid.UUID | None = None,
is_system_admin: bool = False,
cursor: str | None = None,
) -> dict:
"""List contacts with pagination, FTS search, type/folder filter, sorting.
Applies row-level visibility filter based on ownership and entity_permissions.
Keyset-Pagination: If ``cursor`` is provided (a contact UUID), results are
filtered to ``id > cursor`` instead of using OFFSET. This is much faster
for large datasets. When ``cursor`` is not provided, classic page/page_size
offset pagination is used (backward compatible).
"""
from app.core.visibility import apply_visibility_filter
@@ -176,6 +182,11 @@ async def list_contacts(
Contact.search_tsv.op("@@")(func.plainto_tsquery("german", search))
)
# Keyset-Pagination: filter by cursor if provided
use_keyset = cursor is not None and sort_by == "id" and sort_order == "asc"
if use_keyset:
base = base.where(Contact.id > uuid.UUID(cursor))
# Count
count_q = select(func.count()).select_from(base.subquery())
total = (await db.execute(count_q)).scalar() or 0
@@ -187,8 +198,11 @@ async def list_contacts(
base = base.order_by(sort_col)
# Paginate
offset = (page - 1) * page_size
base = base.offset(offset).limit(page_size)
if use_keyset:
base = base.limit(page_size)
else:
offset = (page - 1) * page_size
base = base.offset(offset).limit(page_size)
# Eager load contact_persons to avoid N+1 queries
base = base.options(selectinload(Contact.contact_persons))
@@ -196,11 +210,17 @@ async def list_contacts(
result = await db.execute(base)
contacts = result.scalars().all()
# Next cursor for keyset pagination
next_cursor = None
if use_keyset and len(contacts) == page_size and contacts:
next_cursor = str(contacts[-1].id)
return {
"items": [_serialize_contact(c) for c in contacts],
"total": total,
"page": page,
"page_size": page_size,
"next_cursor": next_cursor,
}