RLS y auditoría

RLS: filtrado por fila

RLS aplica un predicado a la consulta, resolviendo por su cuenta el camino de relaciones hasta el modelo objetivo. Es el mecanismo para que cada usuario vea solo lo suyo sin repetir el mismo WHERE en cada consulta.

from database.public import RLS

regla = RLS(
    target_model=Empresa,      # el modelo sobre el que se filtra
    target_column='id',        # su columna
    values=[1, 2],             # los valores permitidos
    operator='in',             # cómo se comparan
)

posts = public_sync_api.post.find_many(rls=regla)          # una regla
posts = public_sync_api.post.find_many(rls=[regla, otra])  # varias, en AND

Post no tiene por qué apuntar a Empresa directamente: si hay un camino de relaciones (Post → Usuario → Empresa), tai-sql lo encuentra y lo aplica con los JOIN necesarios. Los caminos se cachean, así que el coste de resolverlos se paga una vez.

Una regla sin camino hasta el modelo se ignora

Si no existe ninguna cadena de relaciones entre el modelo consultado y el target_model, la regla no se aplica y la consulta devuelve todo. No es un error: es la diferencia entre «esta regla no habla de esta tabla» y «esta tabla no debe verse». Si una entidad tiene que quedar restringida, comprueba que hay relación hasta el modelo objetivo.

En cambio, si la regla sí aplica pero su predicado no se puede traducir a SQL, la consulta deniega: ignorarla habría devuelto filas de más.

Operadores

in (por defecto), eq, ne, gt, ge, lt, le, like, ilike.

Dónde se puede usar

En todas las lecturas y agregaciones: find, find_many, count, exists, agg, sum, mean, max, min y las búsquedas vectoriales.

RLS no protege las escrituras

create, update, delete y sus variantes masivas no aceptan rls. Si necesitas restringir quién puede escribir qué, eso va en tu capa de aplicación o en un trigger que valide y aborte.

No es Row-Level Security de PostgreSQL

Pese al nombre, esto lo impone el cliente generado añadiendo condiciones a la consulta, no la base de datos. Quien tenga la URL de conexión puede saltárselo. Es una capa de conveniencia para tu aplicación, no un control de seguridad del motor.

Un ejemplo completo

def contexto_del_usuario(usuario) -> RLS:
    return RLS(target_model=Empresa, target_column='id',
               values=usuario.empresas_permitidas)

@app.get('/posts')
def listar(usuario = Depends(auth)):
    return public_sync_api.post.find_many(
        rls=contexto_del_usuario(usuario),
        limit=50,
        includes=['autor'],
    )

Auditoría: quién escribe

Cuando el schema declara operative_fields=True, el cliente rellena created_by y updated_by con el usuario del contexto:

from database.public import set_username, get_username, username_context

set_username('alice')                     # thread-safe y async-safe (ContextVar)
public_sync_api.usuario.create(...)       # created_by = 'alice'

with username_context('system'):          # temporal: al salir se restaura el anterior
    public_sync_api.usuario.create(...)   # created_by = 'system'

get_username()                            # 'alice'

Al ser un ContextVar, cada hilo y cada tarea asíncrona mantienen su propio valor. En una aplicación web, set_username() en un middleware hace que cada request use el suyo:

@app.middleware('http')
async def con_usuario(request, call_next):
    set_username(request.state.user.email)
    return await call_next(request)

El valor por defecto, si nadie lo establece, es 'triplealpha'.