Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
Here we parse some search tokens diffrently to enable filtering:
GSE\d+ and GPL\d+ filter by specific serie or platform
tag=\w+ filters by tag
valid selects validated annotations
"""
options = DatatableOptions._normalize_options(self, query, options)
# Try normally named field
if not options['search']:
options['search'] = query.get('search', '').strip()
filters = group_by(r'^(GSE|GPL|[Tt]ag=|valid|novalid)', options['search'].split())
options['search'] = ' '.join(filters.pop(None, []))
filters = walk_keys(str.lower, filters)
filters['tag'] = lmap(r'^[Tt]ag=(.*)', filters.pop('tag=', []))
options['filters'] = filters
return options
def object(self, children):
required = [
get_first_key(c) for c in children if c != ELLIPSIS and c["required"]
]
children = lmap(lambda o: omit(o, ["required"]), children)
properties = merge(*children)
# if ELLIPSIS in children:
# children.remove(ELLIPSIS)
# if len(children) == 1:
# res = {"type": "object"}
# else:
# res = {"type": "object", "properties": properties}
res = {
"type": "object",
"properties": properties,
"additionalProperties": False,
"required": required,
}
if required:
res.update({"required": required})
def _parse_query(q):
tags, words = lsplit(r'^tag:', q.split())
tags = lmap(r'^tag:(.*)', tags)
return ' '.join(words), tags
def _fetch_all(self):
# This thing appears in Django 1.9.
# In Djangos 1.9 and 1.10 both calls mean the same.
# Starting from Django 1.11 .iterator() uses chunked fetch
# while ._fetch_all() stays with bare _iterable_class.
if hasattr(self, '_iterable_class'):
it = self._iterable_class(self)
else:
it = self.iterator()
self._result_cache = lmap(rcompose(*self._mappers), it)
# Fill in the rest
base._fetch_all(self)
),
)
touch(
f"generated/sdl/general.graphql",
populate_string(general_graphql, dict(searchables=searchables)),
index=True
)
touch(f"generated/sdl/main.graphql", main_graphql_schema, index=True)
implemented_types = []
for typename, type_config in types.items():
type_config = type_config or {}
if not type_config.get("exposed", True):
continue
disambiguations = type_config.get("disambiguations", {})
disambiguations = make_disambiguations_objects(disambiguations)
disambiguations = lmap(add_disambiguations_defaults, disambiguations)
generate_type_boilerplate(
touch=touch,
schema=main_graphql_schema,
collection=type_config.get("collection", ""),
typename=typename,
guards=lmap(add_guards_defaults, type_config.get("guards", [])),
pipeline=type_config.get("pipeline", []),
disambiguations=disambiguations,
)
implemented_types += [typename]
for relation in relations:
toType = relation["to"]
generate_relation_boilerplate(
touch=touch,
schema=main_graphql_schema,
# typename
# collection
# guards
for typename, type_config in types.items():
type_config = type_config or {}
# types with no collection are used only for relations not direct queries
if not type_config.get('exposed', True):
continue
collection = type_config.get('collection', '')
query_name = get_query_name(typename)
pipeline = type_config.get('pipeline', [])
guards = type_config.get('guards', [])
guards = lmap(add_guards_defaults, guards)
disambiguations = type_config.get('disambiguations', {})
disambiguations = make_disambiguations_objects(disambiguations)
disambiguations = lmap(add_disambiguations_defaults, disambiguations)
query_subset = generate_type_sdl(
skema_schema,
guards=guards,
typename=typename,
query_name=query_name
)
touch(f'{base}/generated/sdl/{query_name}.graphql', query_subset)
single_resolver, many_resolver = generate_resolvers(
collection=collection,
disambiguations=disambiguations,
guards=guards,
query_name=query_name,
pipeline=pipeline,
)
touch(f'{base}/generated/resolvers/{get_query_name(typename)}.py', single_resolver)
def low_bounded_range(self, children):
are_floats = any(["." in x for x in children])
if are_floats:
children = lmap(float, children)
else:
children = lmap(int, children)
low, = children
res = {"type": "number", "minimum": low}
if not are_floats:
res.update({"multipleOf": 1.0})
return res
)
touch(f"generated/sdl/main.graphql", main_graphql_schema, index=True)
implemented_types = []
for typename, type_config in types.items():
type_config = type_config or {}
if not type_config.get("exposed", True):
continue
disambiguations = type_config.get("disambiguations", {})
disambiguations = make_disambiguations_objects(disambiguations)
disambiguations = lmap(add_disambiguations_defaults, disambiguations)
generate_type_boilerplate(
touch=touch,
schema=main_graphql_schema,
collection=type_config.get("collection", ""),
typename=typename,
guards=lmap(add_guards_defaults, type_config.get("guards", [])),
pipeline=type_config.get("pipeline", []),
disambiguations=disambiguations,
)
implemented_types += [typename]
for relation in relations:
toType = relation["to"]
generate_relation_boilerplate(
touch=touch,
schema=main_graphql_schema,
fromType=relation["from"],
where_filter=relation["where"],
toType=toType,
pipeline=types[toType].get("pipeline", []),
collection=types[toType].get("collection", []),
relationName=relation.get("field"),
def high_bounded_range(self, children):
are_floats = any(["." in x for x in children])
if are_floats:
children = lmap(float, children)
else:
children = lmap(int, children)
high, = children
res = {"type": "number", "maximum": high}
if not are_floats:
res.update({"multipleOf": 1.0})
return res
hasNext = None
hasPrevious = None
if first:
hasNext = len(nodes) == (first + 1)
nodes = nodes[:-1] if hasNext else nodes
if last:
hasPrevious = len(nodes) == last + 1
nodes = nodes[1:] if hasPrevious else nodes
end_cursor = nodes[-1].get(cursorField) if nodes else None
start_cursor = nodes[0].get(cursorField) if nodes else None
return {
'nodes': nodes,
'edges': lmap(
lambda node: dict(
node=node,
cursor=OUTPUT_COERCERS[scalar_name](node.get(cursorField))
), nodes),
'pageInfo': {
'endCursor': end_cursor and OUTPUT_COERCERS[scalar_name](end_cursor),
'startCursor': start_cursor and OUTPUT_COERCERS[scalar_name](start_cursor),
'hasNextPage': hasNext,
'hasPreviousPage': hasPrevious,
}