-
-
Notifications
You must be signed in to change notification settings - Fork 1.7k
Expand file tree
/
Copy path_util_cy.py
More file actions
127 lines (92 loc) · 3.77 KB
/
_util_cy.py
File metadata and controls
127 lines (92 loc) · 3.77 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
# sql/_util_cy.py
# Copyright (C) 2010-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: disable-error-code="untyped-decorator"
from __future__ import annotations
from typing import Dict
from typing import Literal
from typing import Tuple
from typing import TYPE_CHECKING
from typing import Union
if TYPE_CHECKING:
from .cache_key import CacheConst
from ..engine.interfaces import _CoreSingleExecuteParams
# START GENERATED CYTHON IMPORT
# This section is automatically generated by the script tools/cython_imports.py
try:
# NOTE: the cython compiler needs this "import cython" in the file, it
# can't be only "from sqlalchemy.util import cython" with the fallback
# in that module
import cython
except ModuleNotFoundError:
from sqlalchemy.util import cython
def _is_compiled() -> bool:
"""Utility function to indicate if this module is compiled or not."""
return cython.compiled # type: ignore[no-any-return,unused-ignore]
# END GENERATED CYTHON IMPORT
if cython.compiled:
from cython.cimports.sqlalchemy.util._collections_cy import _get_id
else:
_get_id = id
@cython.cclass
class prefix_anon_map(Dict[str, str]):
"""A map that creates new keys for missing key access.
Considers keys of the form "<ident> <name>" to produce
new symbols "<name>_<index>", where "index" is an incrementing integer
corresponding to <name>.
Inlines the approach taken by :class:`sqlalchemy.util.PopulateDict` which
is otherwise usually used for this type of operation.
"""
def __missing__(self, key: str, /) -> str:
derived: str
value: str
self_dict: dict = self # type: ignore[type-arg]
derived = key.split(" ", 1)[1]
anonymous_counter: int = self_dict.get(derived, 1)
self_dict[derived] = anonymous_counter + 1
value = f"{derived}_{anonymous_counter}"
self_dict[key] = value
return value
_AM_KEY = Union[int, str, "CacheConst"]
_AM_VALUE = Union[int, Literal[True], "_CoreSingleExecuteParams"]
@cython.cclass
class anon_map(Dict[_AM_KEY, _AM_VALUE]):
"""A map that creates new keys for missing key access.
Produces an incrementing sequence given a series of unique keys.
This is similar to the compiler prefix_anon_map class although simpler.
Inlines the approach taken by :class:`sqlalchemy.util.PopulateDict` which
is otherwise usually used for this type of operation.
"""
if cython.compiled:
_index: cython.uint
def __cinit__(self): # type: ignore[no-untyped-def]
self._index = 0
else:
_index: int = 0 # type: ignore[no-redef]
@cython.cfunc
@cython.inline
def _add_missing(self: anon_map, key: _AM_KEY, /) -> int:
val: int = self._index
self._index += 1
self_dict: dict = self # type: ignore[type-arg]
self_dict[key] = val
return val
def get_anon(self: anon_map, obj: object, /) -> Tuple[int, bool]:
self_dict: dict = self # type: ignore[type-arg]
idself: int = _get_id(obj)
if idself in self_dict:
return self_dict[idself], True
else:
return self._add_missing(idself), False
if cython.compiled:
def __getitem__(self: anon_map, key: _AM_KEY, /) -> _AM_VALUE:
self_dict: dict = self # type: ignore[type-arg]
if key in self_dict:
return self_dict[key] # type:ignore[no-any-return]
else:
return self._add_missing(key) # type:ignore[no-any-return]
def __missing__(self: anon_map, key: _AM_KEY, /) -> int:
return self._add_missing(key) # type:ignore[no-any-return]