15909158736

Committed 26 Jun 2025 06:03PM UTC coverage: 81.092% (-0.6%) from 81.702%

Build # 15909158736

Build Type

Pull #850

github

Committed by

web-flow

Commit Message

Merge pull request #875 from bodintsov/feature/share-cleanupgrade-2025-type-annotations

[ENG-7443] Feature/share cleanupgrade 2025 type annotations

Pull Request Pull Request #850: [project][ENG-7225] share clean(up)grade 2025 (milestone 2: upgrade)

Coverage Stats

485 of 534 new or added lines in 63 files covered. (90.82%)

12 existing lines in 5 files now uncovered.

6150 of 7584 relevant lines covered (81.09%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

52.11

/share/util/__init__.py

from collections import OrderedDict
import re


WHITESPACE_RE = r'\s+'


def strip_whitespace(string):
    return re.sub(WHITESPACE_RE, ' ', string).strip()


class InvalidID(Exception):
    def __init__(self, value, message='Invalid ID'):
        super().__init__(value, message)
        self.value = value
        self.message = message


class IDObfuscator:
    NUM = 0xDEADBEEF
    MOD = 10000000000
    MOD_INV = 0x17A991C0F
    # Match HHHHH-HHH-HHH Where H is any hexidecimal digit
    ID_RE = re.compile(r'([0-9A-Fa-f]{2,})([0-9A-Fa-f]{3})-([0-9A-Fa-f]{3})-([0-9A-Fa-f]{3})')

    @classmethod
    def encode(cls, instance) -> str:
        return cls.encode_id(instance.id, instance._meta.model)

    @classmethod
    def encode_id(cls, pk, model):
        from django.contrib.contenttypes.models import ContentType

        model_id = ContentType.objects.get_for_model(model).id
        encoded = '{:09X}'.format(pk * cls.NUM % cls.MOD)
        return '{:02X}{}-{}-{}'.format(model_id, encoded[:3], encoded[3:6], encoded[6:])

    @classmethod
    def decode(cls, id):
        from django.contrib.contenttypes.models import ContentType

        match = cls.ID_RE.match(id)
        if not match:
            raise InvalidID(id)
        model_id, *pks = match.groups()

        try:
            model_class = ContentType.objects.get(pk=int(model_id, 16)).model_class()
        except ContentType.DoesNotExist:
            raise InvalidID(id)

        obj_id = int(''.join(pks), 16) * cls.MOD_INV % cls.MOD

        return (model_class, obj_id)

    @classmethod
    def decode_id(cls, id):
        match = cls.ID_RE.match(id)
        if not match:
            raise InvalidID(id)
        model_id, *pks = match.groups()
        return int(''.join(pks), 16) * cls.MOD_INV % cls.MOD

    @classmethod
    def resolve(cls, id):
        model, pk = cls.decode(id)
        return model.objects.get(pk=pk)

    @classmethod
    def resolver(cls, self, args, context, info):
        return cls.resolve(args.get('id', ''))

    @classmethod
    def load(cls, id, *args):
        model, pk = cls.decode(id)
        try:
            return model.objects.get(pk=pk)
        except model.NotFoundError:
            if args:
                return args[0]
            raise


class BaseJSONAPIMeta:
    @classmethod
    def get_id_from_instance(cls, instance):
        return IDObfuscator.encode(instance)

    @classmethod
    def get_instance_from_id(cls, model_class, id):
        try:
            return IDObfuscator.resolve(id)
        except InvalidID:
            return model_class.objects.get(id=id)


class CyclicalDependency(Exception):
    pass


class TopologicalSorter:
    """Sort a list of nodes topologically, so a node is always preceded by its dependencies.

    Params:
    - `nodes`: Iterable of objects
    - `dependencies`: Callable that takes a single argument (a node) and returns an iterable
        of nodes that should precede it (or keys, if `key` is given)
    - `key`: Callable that takes a single argument (a node) and returns a unique key.
        If omitted, nodes will be compared for equality directly.
    """

    def __init__(self, nodes, dependencies, key=None):
        self.__sorted = []
        self.__nodes = list(nodes)
        self.__visited = set()
        self.__visiting = set()
        self.__dependencies = dependencies
        self.__key = key
        self.__node_map = {key(n): n for n in nodes} if key else None

    def sorted(self):
        if not self.__nodes:
            return self.__sorted

        while self.__nodes:
            n = self.__nodes.pop(0)
            self.__visit(n)

        return self.__sorted

    def __visit(self, node):
        key = self.__key(node) if self.__key else node
        if key in self.__visiting:
            raise CyclicalDependency(key, self.__visiting)

        if key in self.__visited:
            return

        self.__visiting.add(key)
        for k in self.__dependencies(node):
            if k is not None and k is not key:
                self.__visit(self.__get_node(k))

        self.__visited.add(key)
        self.__sorted.append(node)
        self.__visiting.remove(key)

    def __get_node(self, key):
        return self.__node_map[key] if self.__node_map else key


class DictHashingDict:
    # A wrapper around dicts that can have dicts as keys

    def __init__(self):
        self.__inner = {}

    def get(self, key, *args):
        return self.__inner.get(self._hash(key), *args)

    def pop(self, key, *args):
        return self.__inner.pop(self._hash(key), *args)

    def setdefault(self, key, *args):
        return self.__inner.setdefault(self._hash(key), *args)

    def __getitem__(self, key):
        return self.__inner[self._hash(key)]

    def __setitem__(self, key, value):
        self.__inner[self._hash(key)] = value

    def __contains__(self, key):
        return self._hash(key) in self.__inner

    def _hash(self, val):
        if isinstance(val, dict):
            if not isinstance(val, OrderedDict):
                val = tuple((k, self._hash(v)) for k, v in sorted(val.items(), key=lambda x: x[0]))
            else:
                val = tuple((k, self._hash(v)) for k, v in val.items())
        if isinstance(val, (list, tuple)):
            val = tuple(self._hash(v) for v in val)
        return val


def chunked(iterable, size=25, fail_fast=False):
    iterable = iter(iterable)
    try:
        while True:
            chunk = []
            for _ in range(size):
                chunk.append(next(iterable))
            yield chunk
    except StopIteration:
        yield chunk
    except Exception as e:
        if not fail_fast and chunk:
            yield chunk
        raise e


def placeholders(length):
    return ', '.join('%s' for _ in range(length))

1	from collections import OrderedDict	1✔
2	import re	1✔
3
4
5	WHITESPACE_RE = r'\s+'	1✔
6
7
8	def strip_whitespace(string):	1✔
9	return re.sub(WHITESPACE_RE, ' ', string).strip()	×
10
11
12	class InvalidID(Exception):	1✔
13	def __init__(self, value, message='Invalid ID'):	1✔
14	super().__init__(value, message)	1✔
15	self.value = value	1✔
16	self.message = message	1✔
17
18
19	class IDObfuscator:	1✔
20	NUM = 0xDEADBEEF	1✔
21	MOD = 10000000000	1✔
22	MOD_INV = 0x17A991C0F	1✔
23	# Match HHHHH-HHH-HHH Where H is any hexidecimal digit
24	ID_RE = re.compile(r'([0-9A-Fa-f]{2,})([0-9A-Fa-f]{3})-([0-9A-Fa-f]{3})-([0-9A-Fa-f]{3})')	1✔
25
26	@classmethod	1✔
27	def encode(cls, instance) -> str:	1✔
28	return cls.encode_id(instance.id, instance._meta.model)	1✔
29
30	@classmethod	1✔
31	def encode_id(cls, pk, model):	1✔
32	from django.contrib.contenttypes.models import ContentType	1✔
33
34	model_id = ContentType.objects.get_for_model(model).id	1✔
35	encoded = '{:09X}'.format(pk * cls.NUM % cls.MOD)	1✔
36	return '{:02X}{}-{}-{}'.format(model_id, encoded[:3], encoded[3:6], encoded[6:])	1✔
37
38	@classmethod	1✔
39	def decode(cls, id):	1✔
40	from django.contrib.contenttypes.models import ContentType	1✔
41
42	match = cls.ID_RE.match(id)	1✔
43	if not match:	1✔
44	raise InvalidID(id)	1✔
45	model_id, *pks = match.groups()	1✔
46
47	try:	1✔
48	model_class = ContentType.objects.get(pk=int(model_id, 16)).model_class()	1✔
49	except ContentType.DoesNotExist:	×
50	raise InvalidID(id)	×
51
52	obj_id = int(''.join(pks), 16) * cls.MOD_INV % cls.MOD	1✔
53
54	return (model_class, obj_id)	1✔
55
56	@classmethod	1✔
57	def decode_id(cls, id):	1✔
UNCOV 58	match = cls.ID_RE.match(id)	×
UNCOV 59	if not match:	×
60	raise InvalidID(id)	×
UNCOV 61	model_id, *pks = match.groups()	×
UNCOV 62	return int(''.join(pks), 16) * cls.MOD_INV % cls.MOD	×
63
64	@classmethod	1✔
65	def resolve(cls, id):	1✔
66	model, pk = cls.decode(id)	×
67	return model.objects.get(pk=pk)	×
68
69	@classmethod	1✔
70	def resolver(cls, self, args, context, info):	1✔
71	return cls.resolve(args.get('id', ''))	×
72
73	@classmethod	1✔
74	def load(cls, id, *args):	1✔
UNCOV 75	model, pk = cls.decode(id)	×
UNCOV 76	try:	×
UNCOV 77	return model.objects.get(pk=pk)	×
78	except model.NotFoundError:	×
79	if args:	×
80	return args[0]	×
81	raise	×
82
83
84	class BaseJSONAPIMeta:	1✔
85	@classmethod	1✔
86	def get_id_from_instance(cls, instance):	1✔
UNCOV 87	return IDObfuscator.encode(instance)	×
88
89	@classmethod	1✔
90	def get_instance_from_id(cls, model_class, id):	1✔
91	try:	×
92	return IDObfuscator.resolve(id)	×
93	except InvalidID:	×
94	return model_class.objects.get(id=id)	×
95
96
97	class CyclicalDependency(Exception):	1✔
98	pass	1✔
99
100
101	class TopologicalSorter:	1✔
102	"""Sort a list of nodes topologically, so a node is always preceded by its dependencies.
103
104	Params:
105	- `nodes`: Iterable of objects
106	- `dependencies`: Callable that takes a single argument (a node) and returns an iterable
107	of nodes that should precede it (or keys, if `key` is given)
108	- `key`: Callable that takes a single argument (a node) and returns a unique key.
109	If omitted, nodes will be compared for equality directly.
110	"""
111
112	def __init__(self, nodes, dependencies, key=None):	1✔
113	self.__sorted = []	×
114	self.__nodes = list(nodes)	×
115	self.__visited = set()	×
116	self.__visiting = set()	×
117	self.__dependencies = dependencies	×
118	self.__key = key	×
119	self.__node_map = {key(n): n for n in nodes} if key else None	×
120
121	def sorted(self):	1✔
122	if not self.__nodes:	×
123	return self.__sorted	×
124
125	while self.__nodes:	×
126	n = self.__nodes.pop(0)	×
127	self.__visit(n)	×
128
129	return self.__sorted	×
130
131	def __visit(self, node):	1✔
132	key = self.__key(node) if self.__key else node	×
133	if key in self.__visiting:	×
134	raise CyclicalDependency(key, self.__visiting)	×
135
136	if key in self.__visited:	×
137	return	×
138
139	self.__visiting.add(key)	×
140	for k in self.__dependencies(node):	×
141	if k is not None and k is not key:	×
142	self.__visit(self.__get_node(k))	×
143
144	self.__visited.add(key)	×
145	self.__sorted.append(node)	×
146	self.__visiting.remove(key)	×
147
148	def __get_node(self, key):	1✔
149	return self.__node_map[key] if self.__node_map else key	×
150
151
152	class DictHashingDict:	1✔
153	# A wrapper around dicts that can have dicts as keys
154
155	def __init__(self):	1✔
156	self.__inner = {}	×
157
158	def get(self, key, *args):	1✔
159	return self.__inner.get(self._hash(key), *args)	×
160
161	def pop(self, key, *args):	1✔
162	return self.__inner.pop(self._hash(key), *args)	×
163
164	def setdefault(self, key, *args):	1✔
165	return self.__inner.setdefault(self._hash(key), *args)	×
166
167	def __getitem__(self, key):	1✔
168	return self.__inner[self._hash(key)]	×
169
170	def __setitem__(self, key, value):	1✔
171	self.__inner[self._hash(key)] = value	×
172
173	def __contains__(self, key):	1✔
174	return self._hash(key) in self.__inner	×
175
176	def _hash(self, val):	1✔
177	if isinstance(val, dict):	×
178	if not isinstance(val, OrderedDict):	×
179	val = tuple((k, self._hash(v)) for k, v in sorted(val.items(), key=lambda x: x[0]))	×
180	else:
181	val = tuple((k, self._hash(v)) for k, v in val.items())	×
182	if isinstance(val, (list, tuple)):	×
183	val = tuple(self._hash(v) for v in val)	×
184	return val	×
185
186
187	def chunked(iterable, size=25, fail_fast=False):	1✔
188	iterable = iter(iterable)	1✔
189	try:	1✔
190	while True:	1✔
191	chunk = []	1✔
192	for _ in range(size):	1✔
193	chunk.append(next(iterable))	1✔
194	yield chunk	1✔
195	except StopIteration:	1✔
196	yield chunk	1✔
197	except Exception as e:	×
198	if not fail_fast and chunk:	×
199	yield chunk	×
200	raise e	×
201
202
203	def placeholders(length):	1✔
204	return ', '.join('%s' for _ in range(length))	×

CenterForOpenScience / SHARE / 15909158736

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous