Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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:
def field_ops(field: Field) -> Iterator[str]:
yield field.name
for op in field.metadata.get(OPS, ()):
yield f"{field.name}:{op}"
def wrapper(cls):
cls = dataclass(cls)
cls._csv_file = csv_file
cls._csv_id_field = None
cls._csv_parent_field = None
cls._csv_index_field = None
cls._csv_node_list_fields = []
cls._csv_node_fields = []
cls._csv_fields = {}
for field in fields(cls):
csv_type = field.metadata.get(FIELD_METADATA_CSVMODEL_TYPE, None)
if csv_type == 'id':
assert cls._csv_id_field is None, "Already has an ID field"
cls._csv_id_field = field
elif csv_type == 'child_parent_node':
assert cls._csv_parent_field is None, "Already has a parent ID field"
cls._csv_parent_field = field
cls._csv_node_fields.append(field)
elif csv_type == 'child_index':
assert cls._csv_index_field is None, "Already has a list index field"
cls._csv_index_field = field
elif csv_type == 'children_list':
cls._csv_node_list_fields.append(field)
elif csv_type == 'reference':
cls._csv_node_fields.append(field)