async-casbin-sqlmodel-adapter 0.1.5

Async SQLModel Adapter for PyCasbin

Async SQLModel Adapter for PyCasbin

Repo

GitHub

PyPI

Async SQLModel Adapter is the SQLModel adapter for PyCasbin. With this library, Casbin can load policy from SQLModel supported database or save policy to it.

Based on Officially Supported Databases, The current supported databases are:

• PostgreSQL
• MySQL
• SQLite

Installation

pip install async_casbin_sqlmodel_adapter

or

poetry add async-casbin-sqlmodel-adapter

Simple Example

# Stdlib:
import asyncio

# Thirdparty:
import casbin
from async_casbin_sqlmodel_adapter import Adapter
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import Field, SQLModel

engine = create_async_engine("sqlite+aiosqlite:///")


class CasbinRule(SQLModel, table=True):  # type: ignore
    """
    CasbinRule class for SQLModel-based Casbin adapter.
    """

    __tablename__ = "casbin_rule"

    id: int = Field(primary_key=True)
    ptype: str = Field(max_length=255)
    v0: str = Field(max_length=255)
    v1: str = Field(max_length=255)
    v2: str | None = Field(max_length=255, default=None)
    v3: str | None = Field(max_length=255, default=None)
    v4: str | None = Field(max_length=255, default=None)
    v5: str | None = Field(max_length=255, default=None)

    def __str__(self) -> str:
        arr = [self.ptype]
        # pylint: disable=invalid-name
        for v in (self.v0, self.v1, self.v2, self.v3, self.v4, self.v5):
            if v is None:
                break
            arr.append(v)
        return ", ".join(arr)

    def __repr__(self) -> str:
        return f'<CasbinRule {self.id}: "{str(self)}">'


async def main():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    adapter = Adapter(engine)

    e = casbin.AsyncEnforcer("path/to/model.conf", adapter, True)

    sub = "alice"  # the user that wants to access a resource.
    obj = "data1"  # the resource that is going to be accessed.
    act = "read"  # the operation that the user performs on the resource.

    if e.enforce(sub, obj, act):
        # permit alice to read data1async_casbin_sqlmodel_adapter
        pass
    else:
        # deny the request, show an error
        pass


asyncio.run(main())

Getting Help

• PyCasbin

License

This project is licensed under the Apache 2.0 license.