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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
|
from _typeshed import Self
from collections.abc import Mapping
from socket import socket as _socket
from typing import Any, AnyStr, Generic, TypeVar, overload
from .charset import charset_by_id as charset_by_id, charset_by_name as charset_by_name
from .constants import CLIENT as CLIENT, COMMAND as COMMAND, FIELD_TYPE as FIELD_TYPE, SERVER_STATUS as SERVER_STATUS
from .cursors import Cursor
from .util import byte2int as byte2int, int2byte as int2byte
SSL_ENABLED: Any
DEFAULT_USER: Any
DEBUG: Any
DEFAULT_CHARSET: Any
_C = TypeVar("_C", bound=Cursor)
_C2 = TypeVar("_C2", bound=Cursor)
def dump_packet(data): ...
def pack_int24(n): ...
def lenenc_int(i: int) -> bytes: ...
class MysqlPacket:
connection: Any
def __init__(self, data, encoding): ...
def get_all_data(self): ...
def read(self, size): ...
def read_all(self): ...
def advance(self, length): ...
def rewind(self, position: int = ...): ...
def get_bytes(self, position, length: int = ...): ...
def read_string(self) -> bytes: ...
def read_uint8(self) -> Any: ...
def read_uint16(self) -> Any: ...
def read_uint24(self) -> Any: ...
def read_uint32(self) -> Any: ...
def read_uint64(self) -> Any: ...
def read_length_encoded_integer(self) -> int: ...
def read_length_coded_string(self) -> bytes: ...
def read_struct(self, fmt: str) -> tuple[Any, ...]: ...
def is_ok_packet(self) -> bool: ...
def is_eof_packet(self) -> bool: ...
def is_auth_switch_request(self) -> bool: ...
def is_extra_auth_data(self) -> bool: ...
def is_resultset_packet(self) -> bool: ...
def is_load_local_packet(self) -> bool: ...
def is_error_packet(self) -> bool: ...
def check_error(self): ...
def raise_for_error(self) -> None: ...
def dump(self): ...
class FieldDescriptorPacket(MysqlPacket):
def __init__(self, data, encoding): ...
def description(self): ...
def get_column_length(self): ...
class Connection(Generic[_C]):
ssl: Any
host: Any
port: Any
user: Any
password: Any
db: Any
unix_socket: Any
bind_address: Any
charset: Any
use_unicode: Any
client_flag: Any
cursorclass: Any
connect_timeout: Any
messages: Any
encoders: Any
decoders: Any
host_info: Any
sql_mode: Any
init_command: Any
max_allowed_packet: int
server_public_key: bytes
@overload
def __init__(
self: Connection[Cursor], # different between overloads
*,
host: str | None = ...,
user: Any | None = ...,
password: str = ...,
database: Any | None = ...,
port: int = ...,
unix_socket: Any | None = ...,
charset: str = ...,
sql_mode: Any | None = ...,
read_default_file: Any | None = ...,
conv=...,
use_unicode: bool | None = ...,
client_flag: int = ...,
cursorclass: None = ..., # different between overloads
init_command: Any | None = ...,
connect_timeout: int | None = ...,
ssl: Mapping[Any, Any] | None = ...,
ssl_ca=...,
ssl_cert=...,
ssl_disabled=...,
ssl_key=...,
ssl_verify_cert=...,
ssl_verify_identity=...,
read_default_group: Any | None = ...,
compress: Any | None = ...,
named_pipe: Any | None = ...,
autocommit: bool | None = ...,
db: Any | None = ...,
passwd: Any | None = ...,
local_infile: Any | None = ...,
max_allowed_packet: int = ...,
defer_connect: bool | None = ...,
auth_plugin_map: Mapping[Any, Any] | None = ...,
read_timeout: float | None = ...,
write_timeout: float | None = ...,
bind_address: Any | None = ...,
binary_prefix: bool | None = ...,
program_name: Any | None = ...,
server_public_key: bytes | None = ...,
): ...
@overload
def __init__(
self: Connection[_C], # different between overloads
*,
host: str | None = ...,
user: Any | None = ...,
password: str = ...,
database: Any | None = ...,
port: int = ...,
unix_socket: Any | None = ...,
charset: str = ...,
sql_mode: Any | None = ...,
read_default_file: Any | None = ...,
conv=...,
use_unicode: bool | None = ...,
client_flag: int = ...,
cursorclass: type[_C] = ..., # different between overloads
init_command: Any | None = ...,
connect_timeout: int | None = ...,
ssl: Mapping[Any, Any] | None = ...,
ssl_ca=...,
ssl_cert=...,
ssl_disabled=...,
ssl_key=...,
ssl_verify_cert=...,
ssl_verify_identity=...,
read_default_group: Any | None = ...,
compress: Any | None = ...,
named_pipe: Any | None = ...,
autocommit: bool | None = ...,
db: Any | None = ...,
passwd: Any | None = ...,
local_infile: Any | None = ...,
max_allowed_packet: int = ...,
defer_connect: bool | None = ...,
auth_plugin_map: Mapping[Any, Any] | None = ...,
read_timeout: float | None = ...,
write_timeout: float | None = ...,
bind_address: Any | None = ...,
binary_prefix: bool | None = ...,
program_name: Any | None = ...,
server_public_key: bytes | None = ...,
): ...
socket: Any
rfile: Any
wfile: Any
def close(self) -> None: ...
@property
def open(self) -> bool: ...
def autocommit(self, value) -> None: ...
def get_autocommit(self) -> bool: ...
def commit(self) -> None: ...
def begin(self) -> None: ...
def rollback(self) -> None: ...
def select_db(self, db) -> None: ...
def escape(self, obj, mapping: Mapping[Any, Any] | None = ...): ...
def literal(self, obj): ...
def escape_string(self, s: AnyStr) -> AnyStr: ...
@overload
def cursor(self, cursor: None = ...) -> _C: ...
@overload
def cursor(self, cursor: type[_C2]) -> _C2: ...
def query(self, sql, unbuffered: bool = ...) -> int: ...
def next_result(self, unbuffered: bool = ...) -> int: ...
def affected_rows(self): ...
def kill(self, thread_id): ...
def ping(self, reconnect: bool = ...) -> None: ...
def set_charset(self, charset) -> None: ...
def connect(self, sock: _socket | None = ...) -> None: ...
def write_packet(self, payload) -> None: ...
def _read_packet(self, packet_type=...): ...
def insert_id(self): ...
def thread_id(self): ...
def character_set_name(self): ...
def get_host_info(self): ...
def get_proto_info(self): ...
def get_server_info(self): ...
def show_warnings(self): ...
def __enter__(self: Self) -> Self: ...
def __exit__(self, *exc_info: object) -> None: ...
Warning: Any
Error: Any
InterfaceError: Any
DatabaseError: Any
DataError: Any
OperationalError: Any
IntegrityError: Any
InternalError: Any
ProgrammingError: Any
NotSupportedError: Any
class MySQLResult:
connection: Any
affected_rows: Any
insert_id: Any
server_status: Any
warning_count: Any
message: Any
field_count: Any
description: Any
rows: Any
has_next: Any
def __init__(self, connection: Connection[Any]) -> None: ...
first_packet: Any
def read(self) -> None: ...
def init_unbuffered_query(self) -> None: ...
class LoadLocalFile:
filename: Any
connection: Connection[Any]
def __init__(self, filename: Any, connection: Connection[Any]) -> None: ...
def send_data(self) -> None: ...
|
