SQLAlchemy adapter for generating queries with Cerbos: an open core, language-agnostic, scalable authorization solution
Project description
Cerbos + SQLAlchemy Adapter
An adapter library that takes a Cerbos Query Plan (PlanResources API) response and converts it into a SQLAlchemy Select instance. This is designed to work alongside a project using the Cerbos Python SDK.
The following conditions are supported: and
, or
, not
, eq
, ne
, lt
, gt
, le
(lte
), ge
(gte
) and in
. Other operators (eg math operators) can be implemented programatically, and attached to the query object via the query.where(...)
API.
Requirements
- Cerbos > v0.16
- SQLAlchemy >= 1.4 / 2.0
Usage
pip install cerbos-sqlalchemy
from cerbos.sdk.client import CerbosClient
from cerbos.sdk.model import Principal, ResourceDesc
from cerbos_sqlalchemy import get_query
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
from sqlalchemy.sql import Select
Base = declarative_base()
class LeaveRequest(Base):
__tablename__ = "leave_request"
id = Column(Integer, primary_key=True)
department = Column(String(225))
geography = Column(String(225))
team = Column(String(225))
priority = Column(Integer)
with CerbosClient(host="http://localhost:3592") as c:
p = Principal(
"john",
roles={"employee"},
policy_version="20210210",
attr={"department": "marketing", "geography": "GB", "team": "design"},
)
# Get the query plan for "view" action
rd = ResourceDesc("leave_request", policy_version="20210210")
plan = c.plan_resources("view", p, rd)
# the attr_map arg of get_query expects a map[string, InstrumentedAttribute | Column], with cerbos attribute strings mapped to the column/attr instances
attr_map = {
"request.resource.attr.department": LeaveRequest.department, # LeaveRequest.__table__.c.department is also allowed
"request.resource.attr.geography": LeaveRequest.geography,
"request.resource.attr.team": LeaveRequest.team,
"request.resource.attr.priority": LeaveRequest.priority,
}
# `get_query` supports both `Table` instances and ORM entities:
# ORM entity - honouring object level relationships via the sqlalchemy ORM
query: Select = get_query(plan, LeaveRequest, attr_map)
# Alternatively it can generate legacy queries by passing the Table instance
query: Select = get_query(plan, LeaveRequest.__table__, attr_map)
# NOTE: if columns defined within the attr_map originate from more than one table, we need to define a mapping as the optional 4th positional arg to `get_query`.
# The argument is in the form:
# `list[tuple[Table | DeclarativeMeta, BinaryExpression | ColumnOperators]]`
# e.g.:
query: Select = get_query(
plan,
Table1,
{
"request.resource.attr.foo": Table1.foo, # or `Table1.__table__.c.foo`
"request.resource.attr.bar": Table2.bar,
"request.resource.attr.bosh": Table3.bosh,
},
[
(Table2, Table1.table2_id == Table2.id), # or (Table2.__table__, Table1.__table__.c.table2_id == Table2.__table__.c.id)
(Table3, Table1.table3_id == Table3.id),
]
)
# optionally extend the query
query = query.where(LeaveRequest.priority < 5)
# or return a subset of the selected columns (via a new `select`)
# NOTE: this is wise to do as standard, to avoid implicit joins generated by sqla `relationship()` usage, if present
query = query.with_only_columns(
LeaveRequest.department,
LeaveRequest.geography,
)
# Print the compiled query (for debug purposes)
print(query.compile(compile_kwargs={"literal_binds": True}))
Overriding default predicates
By default, the library provides a base set of operators which are widely supported across a range of SQL dialects. However, in some cases, users may wish to override a particular operator for a more idiomatic/optimised alternative for a given database. An example of this could be postgres users preferring to use = ANY
over IN
:
from sqlalchemy.sql.expression import any_
query = get_query(
plan_resource_resp,
some_table,
attr_map={
"request.resource.attr.foo": Table1.foo,
},
# override handler functions in the map below
operator_override_fns={
"in": lambda c, v: c == any_(v),
},
)
The types are as follows:
from sqlalchemy import Column
from sqlalchemy.orm import InstrumentedAttribute
from sqlalchemy.sql.expression import BinaryExpression, ColumnOperators
GenericColumn = Column | InstrumentedAttribute
GenericExpression = BinaryExpression | ColumnOperators
# and the actual map arg to `get_query` ⬇️
OperatorFnMap = dict[str, Callable[[GenericColumn, Any], GenericExpression]]
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for cerbos_sqlalchemy-0.3.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | bc6f3e083e8e8de9d0ef4cebe799f73398dfcae4c4048122d8ab0dab146a39ba |
|
MD5 | 205cf8cca14c60ea3f5fcde38b9f83b7 |
|
BLAKE2b-256 | 34e09945cac1022e63d57de5757ed05730cb75bbd12b2428d4638935e58b2ee0 |