-
Notifications
You must be signed in to change notification settings - Fork 427
/
models.pyi
135 lines (120 loc) · 4.85 KB
/
models.pyi
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
128
129
130
131
132
133
134
135
from .attributes import Attribute
from .exceptions import DoesNotExist as DoesNotExist
from typing import Any, Dict, Generic, Iterable, Iterator, List, Optional, Sequence, Tuple, Type, TypeVar, Text, Union
from pynamodb.connection.table import TableConnection
from pynamodb.expressions.condition import Condition
from pynamodb.pagination import ResultIterator
log: Any
class DefaultMeta: ...
class ResultSet(Iterable):
results: Any
operation: Any
arguments: Any
def __init__(self, results, operation, arguments) -> None: ...
def __iter__(self): ...
class MetaModel(type):
def __init__(self, name: Text, bases: Tuple[type, ...], attrs: Dict[Any, Any]) -> None: ...
_T = TypeVar('_T', bound='Model')
KeyType = Union[Text, bytes, float, int, Tuple]
class Model(metaclass=MetaModel):
DoesNotExist = DoesNotExist
attribute_values: Dict[Text, Any]
_hash_keyname: Optional[str]
_range_keyname: Optional[str]
_connection: Optional[TableConnection]
def __init__(self, hash_key: Optional[KeyType] = ..., range_key: Optional[Any] = ..., **attrs) -> None: ...
@property
def _hash_key(self) -> KeyType: ...
@property
def _range_key(self) -> Optional[Any]: ...
@classmethod
def has_map_or_list_attributes(cls: Type[_T]) -> bool: ...
@classmethod
def batch_get(cls: Type[_T], items: Iterable[Union[KeyType, Iterable[KeyType]]], consistent_read: Optional[bool] = ..., attributes_to_get: Optional[Sequence[Text]] = ...) -> Iterator[_T]: ...
@classmethod
def batch_write(cls: Type[_T], auto_commit: bool = ...) -> BatchWrite[_T]: ...
def delete(self, condition: Optional[Any] = ...) -> Any: ...
def update(self, actions: List[Any], condition: Optional[Condition] = ...) -> Any: ...
def save(self, condition: Optional[Condition] = ...) -> Dict[str, Any]: ...
def refresh(self, consistent_read: bool = ...): ...
@classmethod
def get(cls: Type[_T], hash_key: KeyType, range_key: Optional[KeyType] = ..., consistent_read: bool = ...) -> _T: ...
@classmethod
def from_raw_data(cls: Type[_T], data) -> _T: ...
@classmethod
def count(
cls: Type[_T],
hash_key: Optional[KeyType] = ...,
range_key_condition: Optional[Condition] = ...,
filter_condition: Optional[Condition] = ...,
consistent_read: bool = ...,
index_name: Optional[Text] = ...,
limit: Optional[int] = ...,
rate_limit: Optional[float] = ...,
) -> int: ...
@classmethod
def query(
cls: Type[_T],
hash_key: KeyType,
range_key_condition: Optional[Condition] = ...,
filter_condition: Optional[Condition] = ...,
consistent_read: bool = ...,
index_name: Optional[Text] = ...,
scan_index_forward: Optional[Any] = ...,
limit: Optional[int] = ...,
last_evaluated_key: Optional[Dict[Text, Dict[Text, Any]]] = ...,
attributes_to_get: Optional[Iterable[Text]] = ...,
page_size: Optional[int] = ...,
rate_limit: Optional[float] = ...,
) -> ResultIterator[_T]: ...
@classmethod
def scan(
cls: Type[_T],
filter_condition: Optional[Condition] = ...,
segment: Optional[int] = ...,
total_segments: Optional[int] = ...,
limit: Optional[int] = ...,
last_evaluated_key: Optional[Dict[str, Dict[str, Any]]] = ...,
page_size: Optional[int] = ...,
rate_limit: Optional[float] = ...,
) -> ResultIterator[_T]: ...
@classmethod
def exists(cls: Type[_T]) -> bool: ...
@classmethod
def delete_table(cls): ...
@classmethod
def describe_table(cls): ...
@classmethod
def create_table(cls: Type[_T], wait: bool = ..., read_capacity_units: Optional[Any] = ..., write_capacity_units: Optional[Any] = ...): ...
@classmethod
def dumps(cls): ...
@classmethod
def dump(cls, filename): ...
@classmethod
def loads(cls, data): ...
@classmethod
def load(cls, filename): ...
@classmethod
def add_throttle_record(cls, records): ...
@classmethod
def get_throttle(cls): ...
@classmethod
def get_attributes(cls) -> Dict[str, Attribute]: ...
@classmethod
def _get_attributes(cls) -> Dict[str, Attribute]: ...
@classmethod
def _get_connection(cls) -> TableConnection: ...
class ModelContextManager(Generic[_T]):
model: Type[_T]
auto_commit: bool
max_operations: int
pending_operations: List[Dict[Text, Any]]
def __init__(self, model: Type[_T], auto_commit: bool = ...) -> None: ...
def __enter__(self) -> ModelContextManager[_T]: ...
class BatchWrite(Generic[_T], ModelContextManager[_T]):
def save(self, put_item: _T) -> None: ...
def delete(self, del_item: _T) -> None: ...
def __enter__(self) -> BatchWrite[_T]: ...
def __exit__(self, exc_type, exc_val, exc_tb) -> None: ...
pending_operations: Any
def commit(self) -> None: ...