<commit_before>import graphene
from django.db.models import Q
from django.utils import timezone
from graphene_django.registry import get_global_registry
from graphql.error import GraphQLError
from graphql_relay import from_global_id

from .core.enums import PermissionEnum, ReportingPeriod
from .core.types import PermissionDisplay

registry = get_global_registry()


def get_database_id(info, node_id, only_type):
    """Get a database ID from a node ID of given type."""
    _type, _id = graphene.relay.Node.from_global_id(node_id)
    graphene_type = info.schema.get_type(_type).graphene_type
    if graphene_type != only_type:
        raise AssertionError('Must receive a %s id.' % only_type._meta.name)
    return _id


def get_nodes(ids, graphene_type=None):
    """Return a list of nodes.

    If the `graphene_type` argument is provided, the IDs will be validated
    against this type. If the type was not provided, it will be looked up in
    the Graphene's registry. Raises an error if not all IDs are of the same
    type.
    """
    pks = []
    types = []
    invalid_ids = []
    error_msg = "Could not resolve to a nodes with the global id list of '%s'."
    for graphql_id in ids:
        if graphql_id:
            try:
                _type, _id = from_global_id(graphql_id)
            except Exception:
                invalid_ids.append(graphql_id)
            else:
                if graphene_type:
                    assert str(graphene_type) == _type, (
                        'Must receive an {} id.').format(
                            graphene_type._meta.name)
                pks.append(_id)
                types.append(_type)
    if invalid_ids:
        raise GraphQLError(
            error_msg % invalid_ids)

    # If `graphene_type` was not provided, check if all resolved types are
    # the same. This prevents from accidentally mismatching IDs of different
    # types.
    if types and not graphene_type:
        assert len(set(types)) == 1, 'Received IDs of more than one type.'
        # get type by name
        type_name = types[0]
        for model, _type in registry._registry.items():
            if _type._meta.name == type_name:
                graphene_type = _type
                break

    nodes = list(graphene_type._meta.model.objects.filter(pk__in=pks))
    if not nodes:
        raise GraphQLError(
            error_msg % ids)
    nodes_pk_list = [str(node.pk) for node in nodes]
    for pk in pks:
        assert pk in nodes_pk_list, (
            'There is no node of type {} with pk {}'.format(_type, pk))
    return nodes


def filter_by_query_param(queryset, query, search_fields):
    """Filter queryset according to given parameters.

    Keyword arguments:
    queryset - queryset to be filtered
    query - search string
    search_fields - fields considered in filtering
    """
    if query:
        query_by = {
            '{0}__{1}'.format(
                field, 'icontains'): query for field in search_fields}
        query_objects = Q()
        for q in query_by:
            query_objects |= Q(**{q: query_by[q]})
        return queryset.filter(query_objects).distinct()
    return queryset


def reporting_period_to_date(period):
    now = timezone.now()<commit_msg>Preserve orders from ids for get_nodes<commit_after>import graphene
from django.db.models import Q
from django.utils import timezone
from graphene_django.registry import get_global_registry
from graphql.error import GraphQLError
from graphql_relay import from_global_id

from .core.enums import PermissionEnum, ReportingPeriod
from .core.types import PermissionDisplay

registry = get_global_registry()


def get_database_id(info, node_id, only_type):
    """Get a database ID from a node ID of given type."""
    _type, _id = graphene.relay.Node.from_global_id(node_id)
    graphene_type = info.schema.get_type(_type).graphene_type
    if graphene_type != only_type:
        raise AssertionError('Must receive a %s id.' % only_type._meta.name)
    return _id


def get_nodes(ids, graphene_type=None):
    """Return a list of nodes.

    If the `graphene_type` argument is provided, the IDs will be validated
    against this type. If the type was not provided, it will be looked up in
    the Graphene's registry. Raises an error if not all IDs are of the same
    type.
    """
    pks = []
    types = []
    invalid_ids = []
    error_msg = "Could not resolve to a nodes with the global id list of '%s'."
    for graphql_id in ids:
        if graphql_id:
            try:
                _type, _id = from_global_id(graphql_id)
            except Exception:
                invalid_ids.append(graphql_id)
            else:
                if graphene_type:
                    assert str(graphene_type) == _type, (
                        'Must receive an {} id.').format(
                            graphene_type._meta.name)
                pks.append(_id)
                types.append(_type)
    if invalid_ids:
        raise GraphQLError(
            error_msg % invalid_ids)

    # If `graphene_type` was not provided, check if all resolved types are
    # the same. This prevents from accidentally mismatching IDs of different
    # types.
    if types and not graphene_type:
        assert len(set(types)) == 1, 'Received IDs of more than one type.'
        # get type by name
        type_name = types[0]
        for model, _type in registry._registry.items():
            if _type._meta.name == type_name:
                graphene_type = _type
                break

    nodes = list(graphene_type._meta.model.objects.filter(pk__in=pks))
    nodes.sort(key=lambda e: pks.index(str(e.pk)))  # preserve order in pks
    if not nodes:
        raise GraphQLError(
            error_msg % ids)
    nodes_pk_list = [str(node.pk) for node in nodes]
    for pk in pks:
        assert pk in nodes_pk_list, (
            'There is no node of type {} with pk {}'.format(_type, pk))
    return nodes


def filter_by_query_param(queryset, query, search_fields):
    """Filter queryset according to given parameters.

    Keyword arguments:
    queryset - queryset to be filtered
    query - search string
    search_fields - fields considered in filtering
    """
    if query:
        query_by = {
            '{0}__{1}'.format(
                field, 'icontains'): query for field in search_fields}
        query_objects = Q()
        for q in query_by:
            query_objects |= Q(**{q: query_by[q]})
        return queryset.filter(query_objects).distinct()
    return queryset


def reporting_period_to_date(period):
    now = timezone.now()
{'fraction_non_alphanumeric': 0.0803639120545868, 'fraction_numerical': 0.0012130401819560274, 'mean_word_length': 2.3895169578622815, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3184181', 'n_tokens_mistral': 2012, 'n_tokens_neox': 1903, 'n_words': 621}