smorest_crud package

Module contents

class smorest_crud.AccessControlQuery(entities, session=None)[source]

Bases: flask_sqlalchemy.BaseQuery

Base query class to use for access restriction.

query_for_user(user)[source]

Access control query for the given user instance.

Return type

Type[AccessControlQuery]

class smorest_crud.AccessControlUser(*args, **kwds)[source]

Bases: Generic[smorest_crud.access_control.models.T]

A model mixin to implement access checks for a given model/user.

Required on all models for views with access checks enabled.

Example:

class PetQuery(AccessControlQuery):
    def query_for_user(self, user) -> "PetQuery":
        return self.filter_by(owner=user)

class Pet(Model, AccessControlUser):
    query_class = PetQuery

    def user_can_read(self, user) -> bool:
        return self.user_can_write(user) or self.owner.id == user.id

    def user_can_write(self, user) -> bool:
        return user.is_admin  # only administrators can edit pets
classmethod get_for_user_or_404(user, id_value)[source]

Get instance by key if user allowed to read. :type user: Type[~T] :param user: user instance to check access for :type id_value: Union[str, int] :param id_value: value of the key attribute for filtering

Return type

~T

query_class: Type[smorest_crud.access_control.models.AccessControlQuery]
classmethod query_for_user(user)[source]

Filter list of items for user, or None if disallowed.

Return type

Optional[AccessControlQuery]

user_can_create(user, args)[source]

Check if user is allowed to create.

Return type

bool

user_can_read(user)[source]

Check if user is allowed to access this object at all.

Defaults to calling self.user_can_write(user).

Return type

bool

user_can_write(user)[source]

Check if user can make any modifications to this object (update, delete).

Return type

bool

class smorest_crud.CRUD(app=None)[source]

Bases: object

Flask extension to enable CRUD REST functionality.

Sample full app configuration:

from smorest_crud import CRUD
from flask_jwt_extended import JWTManager, get_current_user

app = Flask()
JWTManager(app)
CRUD(app)

app.config.update(
    CRUD_GET_USER=get_current_user,
    CRUD_ACCESS_CHECKS_ENABLED=True,
    SECRET_KEY="wnt2die",
    CRUD_DEFAULT_KEY_COLUMN="extid",
)
access_control_enabled: bool
app: flask.app.Flask
db: flask_sqlalchemy.SQLAlchemy
get_user: Optional[Callable]
init_app(app, identity_handler=None)[source]
key_attr: str = 'id'
class smorest_crud.CollectionView[source]

Bases: smorest_crud.view.CRUDView

API view that can manage listing items in a collection or creating a new item.

Example:

from flask_smorest import Blueprint
from smorest_crud import CollectionView

pet_blp = Blueprint("pets", "pets", url_prefix="/pet")

@pet_blp.route("")
class PetCollection(CollectionView):
    model = Pet
    prefetch = [Pet.human, (Pet.human, Human.cars)]  # joinedload
    access_checks_enabled = False

    create_enabled = True
    list_enabled = True

    @pet_blp.response(PetSchema(many=True))
    def get(self):
        query = super().get()
        return query.filter_by(name='mischa')

    @pet_blp.arguments(PetSchema)
    @pet_blp.response(PetSchema(many=True))
    def post(self, args):
        return super().post(args)
create_enabled: bool = False

Enable POST.

get()[source]

List collection.

Return type

BaseQuery

Returns

query or iterable of `Model`s.

list_enabled: bool = False

Enable GET.

methods = {'GET', 'POST'}
post(args=None)[source]

Create new model.

Parameters

args – Deserialized schema args.

Returns

Newly-created model.

prefetch: Iterable[sqlalchemy.orm.relationships.RelationshipProperty] = []

List of relationships to prefetch when listing.

class smorest_crud.ResourceView[source]

Bases: smorest_crud.view.CRUDView

Operations to perform on an item, identified in the URL route by a key.

GET /pet/42 – Fetch pet 42.

PATCH /pet/42 – Update pet 42.

DELETE /pet/42 – Delete pet 42.

Example:

from flask_smorest import Blueprint
from smorest_crud import ResourceView

pet_blp = Blueprint("pets", "pets", url_prefix="/pet")

@pet_blp.route("/<int:pk>")
class PetResource(ResourceView):
    model = Pet

    access_checks_enabled = True
    get_enabled = True
    update_enabled = True
    delete_enabled = True

    @pet_blp.response(PetSchema)
    def get(self, pk):
        return super().get(pk)

    @pet_blp.arguments(PetSchema)
    @pet_blp.response(PetSchema)
    def patch(self, args, pk):
        return super().patch(args, pk)

    @pet_blp.response(PetSchema)
    def delete(self, pk):
        return super().delete(pk)
delete(pk)[source]

Delete model.

Parameters

pk – Primary key identifier.

Return type

BaseQuery

delete_enabled: bool = False

Enable DELETE.

get(pk)[source]

Retreieve model by primary key.

Parameters

pk – Primary key identifier.

Return type

BaseQuery

get_enabled: bool = False

Enable GET.

methods = {'DELETE', 'GET', 'PATCH'}
patch(args=None, pk=None)[source]

Update model.

Parameters
  • args – Deserialized request model args.

  • pk – Primary key identifier.

Return type

BaseQuery

Returns

Updated model.

update_enabled: bool = False

Enable PATCH.

smorest_crud.get_for_current_user_or_404(model, id_value)[source]

Get an object by unique column and check if the current user can read it. :type model: Type[~T] :param model: date base model of the instance :type id_value: Union[str, int] :param id_value: the id value of the interested instance

Return type

Optional[~T]

smorest_crud.query_for_current_user(model)[source]

Get query for the current authorized user using access checks. :type model: Type[~T] :param model: date base model of the instance

Return type

AccessControlQuery