Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
).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(
[
m.Book.id.label("id"),
m.Book.title.label("title"),
m.Book.description.label("description"),
m.Book.store_id.label("store"),
def _to_update(self, data: list) -> list:
"""
Prepare resource data for SQLAlchemy update query
"""
to_update_list = []
for data_line in data:
to_update = {}
for fn, v in data_line.items():
f = self.fields[fn]
if isinstance(f, ToMany):
continue
model_field = f.metadata.get("model_field")
if not model_field:
continue
to_update[model_field.key] = v
to_update_list.append(to_update)
return to_update_list
def read__query(self):
fields_to_select = {}
to_group_by = []
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)
def _to_create(self, data: dict) -> dict:
"""
Prepare resource data for SQLAlchemy create query
"""
to_create = {}
for fn, v in data.items():
f = self.fields[fn]
if isinstance(f, ToMany):
continue
model_field = f.metadata["model_field"]
to_create[model_field.key] = v
return to_create