How to use the awokado.custom_fields.ToMany function in awokado

To help you get started, we’ve selected a few awokado examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github 5783354 / awokado / tests / test_app / resources / book.py View on Github external
).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"),
github 5783354 / awokado / awokado / resource.py View on Github external
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
github 5783354 / awokado / awokado / request.py View on Github external
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)
github 5783354 / awokado / awokado / resource.py View on Github external
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