80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
"""OrgScopedQuery helper: central tenant-isolation layer (R-1 mitigation).
|
|
|
|
Every business query passes through this helper so that:
|
|
- `org_id` is always applied (multi-tenant isolation)
|
|
- `deleted_at IS NULL` is applied by default (soft-delete)
|
|
- R-1 risk is mitigated: no query accidentally crosses tenant boundaries
|
|
|
|
In v1 (single-tenant) org_id defaults to 1. In v2 (multi-tenant) the
|
|
router layer passes `current_user.org_id`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
class OrgScopedQuery:
|
|
"""Zentraler Query-Helper, der jede Query mit org_id filtert.
|
|
|
|
v1: org_id = 1 (Single-Tenant-Default)
|
|
v2: org_id = current_user.org_id
|
|
"""
|
|
|
|
def __init__(self, model: type[Any], db: AsyncSession, org_id: int = 1) -> None:
|
|
self.model = model
|
|
self.db = db
|
|
self.org_id = org_id
|
|
|
|
async def get(self, id: int) -> Any:
|
|
"""Fetch a single record by id, scoped to org and not soft-deleted."""
|
|
result = await self.db.execute(
|
|
select(self.model).where(
|
|
self.model.id == id,
|
|
self.model.org_id == self.org_id,
|
|
self.model.deleted_at.is_(None),
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list(
|
|
self,
|
|
skip: int = 0,
|
|
limit: int = 20,
|
|
order_by: Optional[Any] = None,
|
|
**filters: Any,
|
|
) -> list[Any]:
|
|
"""List records scoped to org, with optional filters and pagination."""
|
|
stmt = select(self.model).where(
|
|
self.model.org_id == self.org_id,
|
|
self.model.deleted_at.is_(None),
|
|
)
|
|
for field, value in filters.items():
|
|
if value is not None:
|
|
stmt = stmt.where(getattr(self.model, field) == value)
|
|
if order_by is not None:
|
|
stmt = stmt.order_by(order_by)
|
|
stmt = stmt.offset(skip).limit(limit)
|
|
result = await self.db.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def count(self, **filters: Any) -> int:
|
|
"""Count records scoped to org, with optional filters."""
|
|
from sqlalchemy import func as sa_func
|
|
|
|
stmt = select(sa_func.count()).select_from(self.model).where(
|
|
self.model.org_id == self.org_id,
|
|
self.model.deleted_at.is_(None),
|
|
)
|
|
for field, value in filters.items():
|
|
if value is not None:
|
|
stmt = stmt.where(getattr(self.model, field) == value)
|
|
result = await self.db.execute(stmt)
|
|
return int(result.scalar_one())
|
|
|
|
|
|
__all__ = ["OrgScopedQuery"]
|