Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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",
)
tags = custom_fields.ToMany(
fields.Int(), resource="tag", model_field=m.M2M_Book_Tag.c.tag_id
)
def get_by_author_ids(
self, session, ctx: ReadContext, field: sa.Column = None
):
authors = sa.func.array_remove(
sa.func.array_agg(m.Author.id), None
).label("authors")
q = (
sa.select(
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",
)
tags = custom_fields.ToMany(
fields.Int(), resource="tag", model_field=m.M2M_Book_Tag.c.tag_id
)
def get_by_author_ids(
self, session, ctx: ReadContext, field: sa.Column = None
):
authors = sa.func.array_remove(
def read__includes(self):
for i in self.include or []:
if "." in i:
raise BadRequest()
elif not isinstance(self.resource.fields.get(i), (ToOne, ToMany)):
raise RelationNotFound(f"Relation <{i}> not found")
related_resource_name = self.resource.fields[i].metadata["resource"]
related_res = self.resource.RESOURCES[related_resource_name]
related_field = self.resource.fields[i].metadata.get("model_field")
method_name = f"get_by_{self.resource.Meta.name.lower()}_ids"
if not hasattr(related_res, method_name):
raise BadRequest(
f"Relation {related_resource_name} doesn't ready yet. "
f"Ask developers to add {method_name} method "
f"to {related_resource_name} resource"
)
related_res_obj = related_res()
related_data = getattr(related_res_obj, method_name)(
for field_name, field in self.resource.fields.items():
model_field = field.metadata.get("model_field")
if field.load_only:
continue
if model_field is None:
raise Exception(
f"{self.resource.Meta.name}.{field_name} field must have "
f"'model_field' argument"
)
if isinstance(field, ToMany):
fields_to_select[field_name] = sa.func.array_remove(
sa.func.array_agg(model_field), None
)
elif isinstance(field, ToOne):
fields_to_select[field_name] = model_field
else:
fields_to_select[field_name] = model_field
if (
isinstance(model_field, InstrumentedAttribute)
and model_field.class_ is not self.resource.Meta.model
):
to_group_by.append(model_field)
q = sa.select([clm.label(lbl) for lbl, clm in fields_to_select.items()])
joins = getattr(self.resource.Meta, "select_from", None)
if joins is not None:
q = q.select_from(joins)
if not self.is_list: