Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import sqlalchemy as sa
from marshmallow import fields
import tests.test_app.models as m
from awokado import custom_fields
from awokado.consts import CREATE, READ, UPDATE, BULK_UPDATE, DELETE
from awokado.meta import ResourceMeta
from awokado.request import ReadContext
from awokado.resource import BaseResource
class BookResource(BaseResource):
Meta = ResourceMeta(
model=m.Book,
name="book",
methods=(CREATE, READ, UPDATE, BULK_UPDATE, DELETE),
select_from=sa.outerjoin(
m.Book, m.M2M_Book_Tag, m.Book.id == m.M2M_Book_Tag.c.book_id
).outerjoin(m.Author, m.Book.author_id == m.Author.id),
)
id = fields.Int(model_field=m.Book.id)
title = fields.String(model_field=m.Book.title, required=True)
description = fields.String(model_field=m.Book.description)
author = custom_fields.ToOne(
resource="author", model_field=m.Book.author_id
)
author_name = fields.Str(model_field=m.Author.first_name, dump_only=True)
store = custom_fields.ToOne(
resource="store",
model_field=m.Book.store_id,
description="Store selling book",
return True
@classmethod
def can_delete(cls, session, user_id: int, obj_ids: list, skip_exc=False):
return True
@classmethod
def _get_read_query(cls, ctx, query: Selectable):
return query
class AuthorResource(BaseResource):
Meta = ResourceMeta(
model=m.Author,
name="author",
methods=(CREATE, READ, UPDATE, BULK_UPDATE, DELETE),
auth=AuthorAuth,
select_from=sa.outerjoin(
m.Author, m.Book, m.Author.id == m.Book.author_id
),
)
id = fields.Int(model_field=m.Author.id)
books = custom_fields.ToMany(
fields.Int(),
resource="book",
model_field=m.Book.id,
description="Authors Books",
)
books_count = fields.Int(
dump_only=True, model_field=sa.func.count(m.Book.id)
)
from marshmallow import fields
import tests.test_app.models as m
from awokado.auth import BaseAuth
from awokado.consts import CREATE, READ, UPDATE, BULK_UPDATE, DELETE
from awokado.meta import ResourceMeta
from awokado.resource import BaseResource
class ForbiddenBookResource(BaseResource):
Meta = ResourceMeta(
model=m.Book,
name="forbidden_book",
methods=(CREATE, READ, UPDATE, BULK_UPDATE, DELETE),
auth=BaseAuth,
)
id = fields.Int(model_field=m.Book.id)
title = fields.String(model_field=m.Book.title, required=True)
description = fields.String(model_field=m.Book.description)
import sqlalchemy as sa
from marshmallow import fields
import tests.test_app.models as m
from awokado import custom_fields
from awokado.consts import CREATE, READ, UPDATE, BULK_UPDATE, DELETE
from awokado.meta import ResourceMeta
from awokado.request import ReadContext
from awokado.resource import BaseResource
class TagResource(BaseResource):
Meta = ResourceMeta(
model=m.Tag,
name="tag",
methods=(CREATE, READ, UPDATE, BULK_UPDATE, DELETE),
disable_total=True,
select_from=sa.outerjoin(
m.Tag, m.M2M_Book_Tag, m.Tag.id == m.M2M_Book_Tag.c.tag_id
).outerjoin(m.Book, m.M2M_Book_Tag.c.book_id == m.Book.id),
)
id = fields.Int(model_field=m.Tag.id)
name = fields.String(model_field=m.Tag.name)
books = custom_fields.ToMany(
fields.Int(), resource="book", model_field=m.M2M_Book_Tag.c.book_id
)
book_titles = fields.List(
fields.Str(),
resource="author",
model_field=sa.func.array_remove(sa.func.array_agg(m.Book.title), None),
)
def validate_update_request(self, req: falcon.Request):
methods = self.Meta.methods
if UPDATE not in methods and BULK_UPDATE not in methods:
raise MethodNotAllowed()
payload = json.load(req.bounded_stream)
data = payload.get(self.Meta.name)
try:
deserialized = self.load(data, partial=True, many=True)
except ValidationError as exc:
raise BadRequest(exc.messages)
req.stream = {self.Meta.name: deserialized}