AdmMabe/py_luggage
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

venv active

Browse Source
This commit is contained in:
Mabe
2026-08-26 12:06:39 +02:00
parent b6737d9968
commit 29861506d4
578 changed files with 209234 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
venv/lib/python3.12/site-packages/mariadb/__init__.py
+103
View File
@@ -0,0 +1,103 @@
'''
MariaDB Connector/Python module enables python programs to access MariaDB and
MySQL databases, using an API which is compliant with the Python DB API 2.0
(PEP-249).
'''
import mariadb
from ._mariadb import (
DataError,
DatabaseError,
Error,
IntegrityError,
InterfaceError,
InternalError,
NotSupportedError,
OperationalError,
PoolError,
ProgrammingError,
Warning,
mariadbapi_version,
_have_asan,
)
from .field import fieldinfo
from mariadb.dbapi20 import * # noqa: F401,F403
from mariadb.connectionpool import * # noqa: F401,F403
from mariadb.cursors import Cursor
from mariadb.release_info import __version__ as __version__
from mariadb.release_info import __version_info__ as __version_info__
from mariadb.release_info import __author__ as __author__
from mariadb.connections import Connection
# disable for now, until tests are in place
# from mariadb.pooling import *
_POOLS = _CONNECTION_POOLS = {}
__all__ = ["DataError", "DatabaseError", "Error", "IntegrityError",
"InterfaceError", "InternalError", "NotSupportedError",
"OperationalError", "PoolError", "ProgrammingError",
"Warning", "Connection", "__version__", "__version_info__",
"__author__", "Cursor", "fieldinfo", "_have_asan"]
def connect(*args, connectionclass=mariadb.connections.Connection, **kwargs):
"""
Creates a MariaDB Connection object.
By default, the standard connectionclass mariadb.connections.Connection
will be created.
Parameter connectionclass specifies a subclass of
mariadb.Connection object. If not specified, default will be used.
This optional parameter was added in version 1.1.0.
Connection parameters are provided as a set of keyword arguments:
- **`host`** - The host name or IP address of the database server. If MariaDB Connector/Python was built with MariaDB Connector/C 3.3, it is also possible to provide a comma separated list of hosts for simple fail over in case of one or more hosts are not available.
- **`user`, `username`** - The username used to authenticate with the database server
- **`password`, `passwd`** - The password of the given user
- **`database`, `db`** - Database (schema) name to use when connecting with the database server
- **`unix_socket`** - The location of the unix socket file to use instead of using an IP port to connect. If socket authentication is enabled, this can also be used in place of a password.
- **`port`** - Port number of the database server. If not specified, the default value of 3306 will be used.
- **`connect_timeout`** - Connect timeout in seconds
- **`read_timeout`** - Read timeout in seconds
- **`write_timeout`** - Write timeout in seconds
- **`local_infile`** - Enables or disables the use of LOAD DATA LOCAL INFILE statements.
- **`compress`** (default: `False`) - Uses the compressed protocol for client server communication. If the server doesn't support compressed protocol, the default protocol will be used.
- **`init_command`** - Command(s) which will be executed when connecting and reconnecting to the database server
- **`default_file`** - Read options from the specified option file. If the file is an empty string, default configuration file(s) will be used
- **`default_group`** - Read options from the specified group
- **`plugin_dir`** - Directory which contains MariaDB client plugins.
- **`reconnect`** - Enables or disables automatic reconnect. Available since version 1.1.4
- **`ssl_key`** - Defines a path to a private key file to use for TLS. This option requires that you use the absolute path, not a relative path. The specified key must be in PEM format
- **`ssl_cert`** - Defines a path to the X509 certificate file to use for TLS. This option requires that you use the absolute path, not a relative path. The X609 certificate must be in PEM format.
- **`ssl_ca`** - Defines a path to a PEM file that should contain one or more X509 certificates for trusted Certificate Authorities (CAs) to use for TLS. This option requires that you use the absolute path, not a relative path.
- **`ssl_capath`** - Defines a path to a directory that contains one or more PEM files that contains one X509 certificate for a trusted Certificate Authority (CA)
- **`ssl_cipher`** - Defines a list of permitted cipher suites to use for TLS
- **`ssl_crlpath`** - Defines a path to a PEM file that should contain one or more revoked X509 certificates to use for TLS. This option requires that you use the absolute path, not a relative path.
- **`ssl_verify_cert`** - Enables server certificate verification.
- **`ssl`** - The connection must use TLS security, or it will fail.
- **`tls_version`** - A comma-separated list (without whitespaces) of TLS versions. Valid versions are TLSv1.0, TLSv1.1,TLSv1.2 and TLSv1.3. Added in version 1.1.7.
- **`autocommit`** (default: `False`) - Specifies the autocommit settings. True will enable autocommit, False will disable it (default).
- **`converter`** - Specifies a conversion dictionary, where keys are FIELD_TYPE values and values are conversion functions
"""
if kwargs:
if "pool_name" in kwargs:
if not kwargs["pool_name"] in mariadb._CONNECTION_POOLS:
pool = mariadb.ConnectionPool(**kwargs)
else:
pool = mariadb._CONNECTION_POOLS[kwargs["pool_name"]]
c = pool.get_connection()
return c
connection = connectionclass(*args, **kwargs)
if not isinstance(connection, mariadb.connections.Connection):
raise mariadb.ProgrammingError("%s is not an instance of "
"mariadb.Connection" % connection)
return connection
client_version_info = tuple(int(x, 10) for x in mariadbapi_version.split('.'))
client_version = client_version_info[0] * 10000 +\
client_version_info[1] * 1000 + client_version_info[2]
venv/lib/python3.12/site-packages/mariadb/_mariadb.cpython-312-x86_64-linux-gnu.so
Executable
BIN
View File
Binary file not shown.
venv/lib/python3.12/site-packages/mariadb/connectionpool.py
+301
View File
@@ -0,0 +1,301 @@
#
# Copyright (C) 2020-2021 Georg Richter and MariaDB Corporation AB
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Library General Public
# License as published by the Free Software Foundation; either
# version 2 of the License, or (at your option) any later version.
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Library General Public License for more details.
# You should have received a copy of the GNU Library General Public
# License along with this library; if not see <http://www.gnu.org/licenses>
# or write to the Free Software Foundation, Inc.,
# 51 Franklin St., Fifth Floor, Boston, MA 02110, USA
#
import mariadb
import _thread
import time
from mariadb.constants import STATUS
MAX_POOL_SIZE = 64
class ConnectionPool(object):
"""
Class defining a pool of database connections
MariaDB Connector/Python supports simple connection pooling.
A connection pool holds a number of open connections and handles thread safety when providing connections to threads.
The size of a connection pool is configurable at creation time, but cannot be changed afterward. The maximum size of a connection pool is limited to 64 connections.
Keyword Arguments:
- **`pool_name`** (``str``) - Name of connection pool
- **`pool_size`** (``int``) - Size of pool. The Maximum allowed number is 64. Default to 5
- **`pool_reset_connection`** (``bool``) - Will reset the connection before returning it to the pool. Default to True.
- **`pool_validation_interval`** (``int``) - Specifies the validation interval in milliseconds after which the status of a connection requested from the pool is checked. A value of 0 means that the status will always be checked. Default to 500 (Added in version 1.1.6)
- **\*\*kwargs** - Optional additional connection arguments, as described in mariadb.connect() method.
"""
def __init__(self, *args, **kwargs):
"""
Creates a connection pool class
:param str pool_name:
Name of connection pool
:param int pool_size:
Size of pool. If not specified, the default value of 5 will be used.
The Maximum allowed number is 64.
:param bool pool_reset_connection:
Will reset the connection before returning it to the pool.
The Default value is True.
:param **kwargs kwargs:
Optional additional connection arguments, as described in
mariadb.connect() method.
"""
self._connections_free = []
self._connections_used = []
self._pool_args = {}
self._conn_args = {}
self._lock_pool = _thread.RLock()
self.__closed = 0
key_words = ["pool_name", "pool_size", "pool_reset_connection",
"pool_validation_interval"]
# check if pool_name was provided
if kwargs and "pool_name" in kwargs:
# check if pool_name already exists
if kwargs["pool_name"] in mariadb._CONNECTION_POOLS:
raise mariadb.ProgrammingError("Pool '%s' already exists"
% kwargs["pool_name"])
else:
raise mariadb.ProgrammingError("No pool name specified")
# save pool keyword arguments
self._pool_args["name"] = kwargs.get("pool_name")
self._pool_args["size"] = int(kwargs.get("pool_size", 5))
self._pool_args["reset_connection"] = \
bool(kwargs.get("pool_reset_connection", True))
self._pool_args["validation_interval"] = \
int(kwargs.get("pool_validation_interval", 500))
# validate pool size (must be in range between 1 and MAX_POOL_SIZE)
if not (0 < self._pool_args["size"] <= MAX_POOL_SIZE):
raise mariadb.ProgrammingError("Pool size must be in range of "
"1 and %s" % MAX_POOL_SIZE)
# store pool and connection arguments
self._conn_args = kwargs.copy()
for key in key_words:
if key in self._conn_args:
del self._conn_args[key]
if len(self._conn_args) > 0:
with self._lock_pool:
# fill connection pool
for i in range(0, self._pool_args["size"]):
try:
connection = mariadb.Connection(**self._conn_args)
except mariadb.Error:
# if an error occurred, close all connections
# and raise exception
for j in reversed(range(0, len(self._connections_free))):
try:
self._connections_free[j].close()
except mariadb.Error:
# connect failed, so we are not
# interested in errors
# from close() method
pass
del self._connections_free[j]
raise
self.add_connection(connection)
# store connection pool in _CONNECTION_POOLS
mariadb._CONNECTION_POOLS[self._pool_args["name"]] = self
def _replace_connection(self, connection):
"""
Removes the given connection and adds a new connection.
"""
if connection:
if connection in self._connections_free:
x = self._connections_free.index(connection)
del self._connections_free[x]
elif connection in self._connections_used:
x = self._connections_used.index(connection)
del self._connections_used[x]
connection._Connection__pool = None
connection.close()
return self.add_connection()
def __repr__(self):
if (self.__closed):
return "<mariadb.connectionPool.ConnectionPool object (closed) "\
"at %s>" % (hex(id(self)),)
else:
return "<mariadb.connectionPool.ConnectionPool object (name=%s) "\
"at %s>" % (self.pool_name, hex(id(self)))
def add_connection(self, connection=None):
"""
Adds a connection object to the connection pool.
In case that the pool doesn’t have a free slot or is not configured,
a PoolError exception will be raised.
"""
if not self._conn_args:
raise mariadb.PoolError("Couldn't get configuration for pool %s" %
self._pool_args["name"])
if (connection is not None and
not isinstance(connection, mariadb.connections.Connection)):
raise mariadb.ProgrammingError("Passed parameter is not a "
"connection object")
if connection is None and len(self._conn_args) == 0:
raise mariadb.PoolError("Can't get configuration for pool %s" %
self._pool_args["name"])
total = len(self._connections_free + self._connections_used)
if total >= self._pool_args["size"]:
raise mariadb.PoolError("Can't add connection to pool %s: "
"No free slot available (%s)." %
(self._pool_args["name"],
total))
with self._lock_pool:
if connection is None:
connection = mariadb.Connection(**self._conn_args)
connection._Connection__pool = self
connection.__last_used = time.perf_counter_ns()
self._connections_free.append(connection)
return connection
def get_connection(self):
"""
Returns a connection from the connection pool or raises a PoolError
exception if a connection is not available.
"""
conn = None
with self._lock_pool:
for i in range(0, len(self._connections_free)):
conn = self._connections_free[i]
dt = (time.perf_counter_ns() - conn.__last_used) / 1000000
if dt > self._pool_args["validation_interval"]:
try:
conn.ping()
except mariadb.Error:
conn = self._replace_connection(conn)
if not conn:
continue
conn._used += 1
self._connections_used.append(conn)
idx = self._connections_free.index(conn)
del self._connections_free[idx]
return conn
raise mariadb.PoolError("No connection available")
def _close_connection(self, connection):
"""
Returns connection to the pool. Internally used
by connection object.
"""
with self._lock_pool:
try:
if self._pool_args["reset_connection"]:
connection.reset()
elif connection.server_status & STATUS.IN_TRANS:
connection.rollback()
except mariadb.Error:
self._replace_connection(connection)
if connection:
if connection in self._connections_used:
x = self._connections_used.index(connection)
del self._connections_used[x]
connection.__last_used = time.perf_counter_ns()
self._connections_free.append(connection)
def set_config(self, **kwargs):
"""
Sets the connection configuration for the connection pool.
For valid connection arguments, check the mariadb.connect() method.
Note: This method doesn't create connections in the pool.
To fill the pool, one has to use add_connection() ḿethod.
"""
self._conn_args = kwargs
def close(self):
"""Closes connection pool and all connections."""
try:
for c in (self._connections_free + self._connections_used):
c._Connection__pool = None
c.close()
finally:
self._connections_free = None
self._connections_used = None
del mariadb._CONNECTION_POOLS[self._pool_args["name"]]
@property
def pool_name(self):
"""Returns the name of the connection pool."""
return self._pool_args["name"]
@property
def pool_size(self):
"""Returns the size of the connection pool."""
return self._pool_args["size"]
@property
def max_size(self):
"Returns the maximum size for connection pools."""
return MAX_POOL_SIZE
@property
def connection_count(self):
"Returns the number of connections in connection pool."""
try:
return len(self._connections_free + self._connections_used)
except Exception:
return 0
@property
def pool_reset_connection(self):
"""
If set to true, the connection will be reset on both client and server
side after .close() method was called
"""
return self._pool_args["reset_connection"]
@pool_reset_connection.setter
def pool_reset_connection(self, reset):
self._pool_args["reset_connection"] = reset
venv/lib/python3.12/site-packages/mariadb/connections.py
+692
View File
@@ -0,0 +1,692 @@
#
# Copyright (C) 2020-2021 Georg Richter and MariaDB Corporation AB
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Library General Public
# License as published by the Free Software Foundation; either
# version 2 of the License, or (at your option) any later version.
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Library General Public License for more details.
# You should have received a copy of the GNU Library General Public
# License along with this library; if not see <http://www.gnu.org/licenses>
# or write to the Free Software Foundation, Inc.,
# 51 Franklin St., Fifth Floor, Boston, MA 02110, USA
#
import mariadb
import socket
import mariadb.cursors
from mariadb.constants import STATUS, TPC_STATE, INFO
from packaging import version
_DEFAULT_CHARSET = "utf8mb4"
_DEFAULT_COLLATION = "utf8mb4_general_ci"
_MAX_TPC_XID_SIZE = 64
class Connection(mariadb._mariadb.connection):
"""
MariaDB Connector/Python Connection Object
Handles the connection to a MariaDB or MySQL database server.
It encapsulates a database session.
Connections are created using the method mariadb.connect()
"""
def _check_closed(self):
if self._closed:
raise mariadb.ProgrammingError("Invalid connection or "
"not connected")
def __init__(self, *args, **kwargs):
"""
Establishes a connection to a database server and returns a connection
object.
"""
self._socket = None
self._used = 0
self._last_executed_statement = None
self._socket = None
self.__pool = None
self.__last_used = 0
self.tpc_state = TPC_STATE.NONE
self._xid = None
autocommit = kwargs.pop("autocommit", False)
reconnect = kwargs.pop("reconnect", False)
self._converter = kwargs.pop("converter", None)
# if host contains a connection string or multiple hosts,
# we need to check if it's supported by Connector/C
if "host" in kwargs:
host = kwargs.get("host")
if version.Version(mariadb.mariadbapi_version) <\
version.Version('3.3.0') and ',' in host:
raise mariadb.ProgrammingError("Host failover list requires "
"MariaDB Connector/C 3.3.0 "
"or newer")
# compatibility feature: if SSL is provided as a dictionary,
# we will map it's content
if "ssl" in kwargs and not isinstance(kwargs["ssl"], bool):
ssl = kwargs.pop("ssl", None)
for key in ["ca", "cert", "capath", "key", "cipher"]:
if key in ssl:
kwargs["ssl_%s" % key] = ssl[key]
kwargs["ssl"] = True
super().__init__(*args, **kwargs)
self.autocommit = autocommit
self.auto_reconnect = reconnect
def cursor(self, cursorclass=mariadb.cursors.Cursor, **kwargs):
"""
Returns a new cursor object for the current connection.
If no cursorclass was specified, a cursor with default mariadb.Cursor class will be created.
Optional keyword parameters:
- **buffered** (default: ``True``) - If disabled, the result will be unbuffered, which means before executing another statement with the same connection, the entire result set must be fetched. Please note that the default was False for MariaDB Connector/Python versions < 1.1.0.
- **dictionary** (default: ``False``) - Return fetch values as dictionary when enabled.
- **named_tuple** (default: ``False``) - Return fetch values as named tuple. This feature exists for compatibility reasons and should be avoided due to possible inconsistency.
- **cursor_type** (default: ``CURSOR.NONE``) - If cursor_type is set to CURSOR.READ_ONLY, a cursor is opened for the statement invoked with cursors execute() method.
- **prepared** (default: ``False``) - When enabled, the cursor will remain in prepared state after the first execute() method was called. Further calls to execute() method will ignore the SQL statement.
- **binary** (default: ``False``) - Always execute statement in MariaDB client/server binary protocol.
In versions prior to 1.1.0 results were unbuffered by default, which means before executing another statement with the same connection, the entire result set must be fetched.
fetch* methods of the cursor class by default return result set values as a tuple, unless dictionary or named_tuple was specified. The latter one exists for compatibility reasons and should be avoided due to possible inconsistency in case two or more fields in a result set have the same name.
If cursor_type is set to CURSOR.READ_ONLY, a cursor is opened for the statement invoked with cursors execute() method.
"""
self._check_closed()
cursor = cursorclass(self, **kwargs)
if not isinstance(cursor, mariadb._mariadb.cursor):
raise mariadb.ProgrammingError("%s is not an instance of "
"mariadb.cursor" % cursor)
return cursor
def close(self):
self._check_closed()
if self._Connection__pool:
self._Connection__pool._close_connection(self)
else:
super().close()
def __enter__(self):
self._check_closed()
"Returns a copy of the connection."
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self._check_closed()
"Closes connection."
self.close()
def commit(self):
"""
Commit any pending transaction to the database.
"""
self._check_closed()
if self.tpc_state > TPC_STATE.NONE:
raise mariadb.ProgrammingError("commit() is not allowed if "
"a TPC transaction is active")
self._execute_command("COMMIT")
self._read_response()
def rollback(self):
"""
Causes the database to roll back to the start of any pending
transaction
Closing a connection without committing the changes first will
cause an implicit rollback to be performed.
Note that rollback() will not work as expected if autocommit mode
was set to True or the storage engine does not support transactions."
"""
self._check_closed()
if self.tpc_state > TPC_STATE.NONE:
raise mariadb.ProgrammingError("rollback() is not allowed if a "
"TPC transaction is active")
self._execute_command("ROLLBACK")
self._read_response()
def kill(self, id: int):
"""
This function is used to ask the server to kill a database connection
specified by the processid parameter.
The connection id can be retrieved by `SHOW PROCESSLIST` SQL command.
"""
self._check_closed()
if not isinstance(id, int):
raise mariadb.ProgrammingError("id must be of type int.")
stmt = "KILL %s" % id
self._execute_command(stmt)
self._read_response()
def begin(self):
"""
Start a new transaction which can be committed by .commit() method,
or canceled by .rollback() method.
"""
self._check_closed()
self._execute_command("BEGIN")
self._read_response()
def select_db(self, new_db: str):
"""
Gets the default database for the current connection.
The default database can also be obtained or changed by database
attribute.
"""
self._check_closed()
self.database = new_db
def get_server_version(self):
"""
Returns a tuple representing the version of the connected server in
the following format: (MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION)
"""
return self.server_version_info
def show_warnings(self):
"""
Shows error, warning and note messages from last executed command.
"""
self._check_closed()
if (not self.warnings):
return None
cursor = self.cursor()
cursor.execute("SHOW WARNINGS")
ret = cursor.fetchall()
del cursor
return ret
class xid(tuple):
"""
xid(format_id: int, global_transaction_id: str, branch_qualifier: str)
Creates a transaction ID object suitable for passing to the .tpc_*()
methods of this connection.
Parameters:
- format_id: Format id. Default to value `0`.
- global_transaction_id: Global transaction qualifier, which must be
unique. The maximum length of the global transaction id is
limited to 64 characters.
- branch_qualifier: Branch qualifier which represents a local
transaction identifier. The maximum length of the branch qualifier
is limited to 64 characters.
"""
def __new__(self, format_id, transaction_id, branch_qualifier):
if not isinstance(format_id, int):
raise mariadb.ProgrammingError("argument 1 must be int, "
"not %s",
type(format_id).__name__)
if not isinstance(transaction_id, str):
raise mariadb.ProgrammingError("argument 2 must be str, "
"not %s",
type(transaction_id).__mane__)
if not isinstance(branch_qualifier, str):
raise mariadb.ProgrammingError("argument 3 must be str, "
"not %s",
type(transaction_id).__name__)
if len(transaction_id) > _MAX_TPC_XID_SIZE:
raise mariadb.ProgrammingError("Maximum length of "
"transaction_id exceeded.")
if len(branch_qualifier) > _MAX_TPC_XID_SIZE:
raise mariadb.ProgrammingError("Maximum length of "
"branch_qualifier exceeded.")
if format_id == 0:
format_id = 1
return super().__new__(self, (format_id,
transaction_id,
branch_qualifier))
def tpc_begin(self, xid):
"""
Parameter:
xid: xid object which was created by .xid() method of connection
class
Begins a TPC transaction with the given transaction ID xid.
This method should be called outside a transaction
(i.e., nothing may have been executed since the last .commit()
or .rollback()).
Furthermore, it is an error to call .commit() or .rollback() within
the TPC transaction. A ProgrammingError is raised if the application
calls .commit() or .rollback() during an active TPC transaction.
"""
self._check_closed()
if type(xid).__name__ != "xid":
raise mariadb.ProgrammingError("argument 1 must be xid "
"not %s", type(xid).__name__)
stmt = "XA BEGIN '%s','%s',%s" % (xid[1], xid[2], xid[0])
try:
self._execute_command(stmt)
self._read_response()
except mariadb.Error:
raise
self.tpc_state = TPC_STATE.XID
self._xid = xid
def tpc_commit(self, xid=None):
"""
Optional parameter:
- xid
: xid object which was created by .xid() method of connection class.
When called with no arguments, .tpc_commit() commits a TPC transaction
previously prepared with .tpc_prepare().
If .tpc_commit() is called prior to .tpc_prepare(), a single phase
commit is performed. A transaction manager may choose to do this if
only a single resource is participating in the global transaction.
When called with a transaction ID xid, the database commits the given
transaction. If an invalid transaction ID is provided,
a ProgrammingError will be raised.
This form should be called outside a transaction, and
is intended for use in recovery.
"""
self._check_closed()
if not xid:
xid = self._xid
if self.tpc_state == TPC_STATE.NONE:
raise mariadb.ProgrammingError("Transaction not started.")
if xid is None and self.tpc_state != TPC_STATE.PREPARE:
raise mariadb.ProgrammingError("Transaction is not prepared.")
if xid and type(xid).__name__ != "xid":
raise mariadb.ProgrammingError("argument 1 must be xid "
"not %s" % type(xid).__name__)
if self.tpc_state < TPC_STATE.PREPARE:
stmt = "XA END '%s','%s',%s" % (xid[1], xid[2], xid[0])
self._execute_command(stmt)
try:
self._read_response()
except mariadb.Error:
self._xid = None
self.tpc_state = TPC_STATE.NONE
raise
stmt = "XA COMMIT '%s','%s',%s" % (xid[1], xid[2], xid[0])
if self.tpc_state < TPC_STATE.PREPARE:
stmt = stmt + " ONE PHASE"
try:
self._execute_command(stmt)
self._read_response()
except mariadb.Error:
self._xid = None
self.tpc_state = TPC_STATE.NONE
raise
# cleanup
self._xid = None
self.tpc_state = TPC_STATE.NONE
def tpc_prepare(self):
"""
Performs the first phase of a transaction started with .tpc_begin().
A ProgrammingError will be raised if this method was called outside
a TPC transaction.
After calling .tpc_prepare(), no statements can be executed until
.tpc_commit() or .tpc_rollback() have been called.
"""
self._check_closed()
if self.tpc_state == TPC_STATE.NONE:
raise mariadb.ProgrammingError("Transaction not started.")
if self.tpc_state == TPC_STATE.PREPARE:
raise mariadb.ProgrammingError("Transaction is already in "
"prepared state.")
xid = self._xid
stmt = "XA END '%s','%s',%s" % (xid[1], xid[2], xid[0])
try:
self._execute_command(stmt)
self._read_response()
except mariadb.Error:
self._xid = None
self.tpc_state = TPC_STATE.NONE
raise
stmt = "XA PREPARE '%s','%s',%s" % (xid[1], xid[2], xid[0])
try:
self._execute_command(stmt)
self._read_response()
except mariadb.Error:
self._xid = None
self.tpc_state = TPC_STATE.NONE
raise
self.tpc_state = TPC_STATE.PREPARE
def tpc_rollback(self, xid=None):
"""
Parameter:
xid: xid object which was created by .xid() method of connection
class
Performs the first phase of a transaction started with .tpc_begin().
A ProgrammingError will be raised if this method outside a TPC
transaction.
After calling .tpc_prepare(), no statements can be executed until
.tpc_commit() or .tpc_rollback() have been called.
"""
self._check_closed()
if self.tpc_state == TPC_STATE.NONE:
raise mariadb.ProgrammingError("Transaction not started.")
if xid and type(xid).__name__ != "xid":
raise mariadb.ProgrammingError("argument 1 must be xid "
"not %s" % type(xid).__name__)
if not xid:
xid = self._xid
if self.tpc_state < TPC_STATE.PREPARE:
stmt = "XA END '%s','%s',%s" % (xid[1], xid[2], xid[0])
self._execute_command(stmt)
try:
self._read_response()
except mariadb.Error:
self._xid = None
self.tpc_state = TPC_STATE.NONE
raise
stmt = "XA ROLLBACK '%s','%s',%s" % (xid[1], xid[2], xid[0])
try:
self._execute_command(stmt)
self._read_response()
except mariadb.Error:
self._xid = None
self.tpc_state = TPC_STATE.NONE
raise
self.tpc_state = TPC_STATE.PREPARE
def tpc_recover(self):
"""
Returns a list of pending transaction IDs suitable for use with
tpc_commit(xid) or .tpc_rollback(xid).
"""
self._check_closed()
cursor = self.cursor()
cursor.execute("XA RECOVER")
result = cursor.fetchall()
del cursor
return result
@property
def tls_peer_cert_info(self):
"""Get peer certificate information."""
if version.Version(mariadb.mariadbapi_version) <\
version.Version('3.4.2'):
return None
self._check_closed()
if self._tls:
return self._mariadb_get_info(INFO.TLS_PEER_CERT_INFO)
return None
@property
def database(self):
"""Get the current database of the connection."""
self._check_closed()
return self._mariadb_get_info(INFO.SCHEMA)
@database.setter
def database(self, schema):
"""Set default database."""
self._check_closed()
try:
self._execute_command("USE %s" % str(schema))
self._read_response()
except mariadb.Error:
raise
@property
def user(self):
"""
Returns the username for the current connection or empty
string if it can't be determined, e.g., when using socket
authentication.
"""
self._check_closed()
return self._mariadb_get_info(INFO.USER)
@property
def character_set(self):
"""
Client character set.
For MariaDB Connector/Python, it is always utf8mb4.
"""
return _DEFAULT_CHARSET
@property
def client_capabilities(self):
"""Client capability flags."""
self._check_closed()
return self._mariadb_get_info(INFO.CLIENT_CAPABILITIES)
@property
def server_capabilities(self):
"""Server capability flags."""
self._check_closed()
return self._mariadb_get_info(INFO.SERVER_CAPABILITIES)
@property
def extended_server_capabilities(self):
"""
Extended server capability flags (only for MariaDB
database servers).
"""
self._check_closed()
return self._mariadb_get_info(INFO.EXTENDED_SERVER_CAPABILITIES)
@property
def server_port(self):
"""
Database server TCP/IP port. This value will be 0 in case of an unix
socket connection.
"""
self._check_closed()
return self._mariadb_get_info(INFO.PORT)
@property
def unix_socket(self):
"""Unix socket name."""
self._check_closed()
return self._mariadb_get_info(INFO.UNIX_SOCKET)
@property
def server_name(self):
"""Name or IP address of database server."""
self._check_closed()
return self._mariadb_get_info(INFO.HOST)
@property
def collation(self):
"""Client character set collation"""
return _DEFAULT_COLLATION
@property
def server_info(self):
"""Server version in alphanumerical format (str)"""
self._check_closed()
return self._mariadb_get_info(INFO.SERVER_VERSION)
@property
def tls_cipher(self):
"""TLS cipher suite if a secure connection is used."""
self._check_closed()
if self._tls:
return self._mariadb_get_info(INFO.SSL_CIPHER)
return None
@property
def tls_version(self):
"""TLS protocol version if a secure connection is used."""
self._check_closed()
if self._tls:
return self._mariadb_get_info(INFO.TLS_VERSION)
return None
@property
def _tls_verify_status(self):
"""Returns the result of the peer certificate verification."""
if version.Version(mariadb.mariadbapi_version) <\
version.Version('3.4.2'):
return None
self._check_closed()
if self._tls:
return self._mariadb_get_info(INFO.TLS_VERIFY_STATUS)
return None
@property
def server_status(self):
"""
Return server status flags
"""
self._check_closed()
return self._mariadb_get_info(INFO.SERVER_STATUS)
@property
def server_version(self):
"""
Server version in numerical format.
The form of the version number is
VERSION_MAJOR * 10000 + VERSION_MINOR * 100 + VERSION_PATCH
"""
self._check_closed()
return self._mariadb_get_info(INFO.SERVER_VERSION_ID)
@property
def server_version_info(self):
"""
Returns numeric version of connected database server in tuple format.
"""
self._check_closed()
version = self.server_version
return (int(version / 10000),
int((version % 10000) / 100),
version % 100)
@property
def autocommit(self):
"""
Toggles autocommit mode on or off for the current database connection.
Autocommit mode only affects operations on transactional table types.
Be aware that rollback() will not work if autocommit mode was switched
on.
By default, autocommit mode is set to False.
"""
self._check_closed()
return bool(self.server_status & STATUS.AUTOCOMMIT)
@autocommit.setter
def autocommit(self, mode):
self._check_closed()
if bool(mode) == self.autocommit:
return
try:
self._execute_command("SET AUTOCOMMIT=%s" % int(mode))
self._read_response()
except mariadb.Error:
raise
@property
def socket(self):
"""Returns the socket used for database connection"""
fno = self._get_socket()
if not self._socket:
self._socket = socket.socket(fileno=fno)
# in case of a possible reconnect, file descriptor has changed
elif fno != self._socket.fileno():
self._socket = socket.socket(fileno=fno)
return self._socket
@property
def open(self):
"""
Returns true if the connection is alive.
A ping command will be sent to the server for this purpose,
which means this function might fail if there are still
non-processed pending result sets.
"""
self._check_closed()
try:
self.ping()
except mariadb.Error:
return False
return True
# Aliases
character_set_name = character_set
@property
def thread_id(self):
"""
Alias for connection_id
"""
self._check_closed()
return self.connection_id
venv/lib/python3.12/site-packages/mariadb/constants/CAPABILITY.py
+38
View File
@@ -0,0 +1,38 @@
'''
MariaDB capability flags.
These flags are used to check the capabilities both of a MariaDB server
or the client applicaion.
Capability flags are defined in module *mariadb.constants.CAPABILIY*
'''
MYSQL = 1 # MariaDB
LONG_PASSWORD = 1 # MySQL
FOUND_ROWS = 2
LONG_FLAG = 4
CONNECT_WITH_DB = 8
NO_SCHEMA = 16
COMPRESS = 32
LOCAL_FILES = 128
IGNORE_SPACE = 256
INTERACTIVE = 1024
SSL = 2048
TRANSACTIONS = 8192
SECURE_CONNECTION = 32768
MULTI_STATEMENTS = 1 << 16
MULTI_RESULTS = 1 << 17
PS_MULTI_RESULTS = 1 << 18
PLUGIN_AUTH = 1 << 19
CONNECT_ATTRS = 1 << 20
CAN_HANDLE_EXPIRED_PASSWORDS = 1 < 22
SESSION_TRACKING = 1 << 23
SSL_VERIFY_SERVER_CERT = 1 << 30
REMEMBER_OPTIONS = 1 << 31
# MariaDB specific capabilities
PROGRESS = 1 << 32
BULK_OPERATIONS = 1 << 34
EXTENDED_METADATA = 1 << 35
CACHE_METDATA = 1 << 36
venv/lib/python3.12/site-packages/mariadb/constants/CLIENT.py
+38
View File
@@ -0,0 +1,38 @@
'''
MariaDB capability flags.
These flags are used to check the capabilities both of a MariaDB server
or the client applicaion.
Capability flags are defined in module *mariadb.constants.CLIENT*
'''
MYSQL = 1 # MariaDB
LONG_PASSWORD = 1 # MySQL
FOUND_ROWS = 2
LONG_FLAG = 4
CONNECT_WITH_DB = 8
NO_SCHEMA = 16
COMPRESS = 32
LOCAL_FILES = 128
IGNORE_SPACE = 256
INTERACTIVE = 1024
SSL = 2048
TRANSACTIONS = 8192
SECURE_CONNECTION = 32768
MULTI_STATEMENTS = 1 << 16
MULTI_RESULTS = 1 << 17
PS_MULTI_RESULTS = 1 << 18
PLUGIN_AUTH = 1 << 19
CONNECT_ATTRS = 1 << 20
CAN_HANDLE_EXPIRED_PASSWORDS = 1 < 22
SESSION_TRACKING = 1 << 23
SSL_VERIFY_SERVER_CERT = 1 << 30
REMEMBER_OPTIONS = 1 << 31
# MariaDB specific capabilities
PROGRESS = 1 << 32
BULK_OPERATIONS = 1 << 34
EXTENDED_METADATA = 1 << 35
CACHE_METDATA = 1 << 36
venv/lib/python3.12/site-packages/mariadb/constants/CURSOR.py
+9
View File
@@ -0,0 +1,9 @@
"""
Cursor constants are used for server side cursors.
Currently only read only cursor is supported.
Cursor constants are defined in module *mariadb.constants.CURSOR*.
"""
NONE = 0
READ_ONLY = 1
venv/lib/python3.12/site-packages/mariadb/constants/ERR.py
+1254
View File
File diff suppressed because it is too large Load Diff
venv/lib/python3.12/site-packages/mariadb/constants/EXT_FIELD_TYPE.py
+20
View File
@@ -0,0 +1,20 @@
"""
MariaDB EXT_FIELD_TYPE Constants
These constants represent the extended field types supported by MariaDB.
Extended field types are defined in module *mariadb.constants.EXT_FIELD_TYPE*
"""
NONE =0
JSON = 1
UUID = 2
INET4 = 3
INET6 = 4
POINT = 5
MULTIPOINT = 6
LINESTRING = 7
MULTILINESTRING = 8
POLYGON = 9
MULTIPOLYGON = 10
GEOMETRYCOLLECTION = 11
venv/lib/python3.12/site-packages/mariadb/constants/FIELD_FLAG.py
+29
View File
@@ -0,0 +1,29 @@
"""MariaDB FIELD_FLAG Constants
These constants represent the various field flags. As an addition
to the DBAPI 2.0 standard (PEP-249) these flags are returned as
eighth element of the cursor description attribute.
Field flags are defined in module *mariadb.constants.FIELD_FLAG*
"""
# Source: mariadb_com.h (MariaDB Connector(C)
NOT_NULL = 1
PRIMARY_KEY = 2
UNIQUE_KEY = 4
MULTIPLE_KEY = 8
BLOB = 16
UNSIGNED = 32
ZEROFILL = 64
BINARY = 128
ENUM = 256
AUTO_INCREMENT = 512
TIMESTAMP = 1024
SET = 2048
NO_DEFAULT = 4096
ON_UPDATE_NOW = 8192
NUMERIC = 32768
PART_OF_KEY = 16384
GROUP = 32768
UNIQUE = 65536
venv/lib/python3.12/site-packages/mariadb/constants/FIELD_TYPE.py
+40
View File
@@ -0,0 +1,40 @@
"""
MariaDB FIELD_TYPE Constants
These constants represent the field types supported by MariaDB.
The field type is returned as second element of cursor description attribute.
Field types are defined in module *mariadb.constants.FIELD_TYPE*
"""
DECIMAL = 0
TINY = 1
SHORT = 2
LONG = 3
FLOAT = 4
DOUBLE = 5
NULL = 6
TIMESTAMP = 7
LONGLONG = 8
INT24 = 9
DATE = 10
TIME = 11
DATETIME = 12
YEAR = 13
NEWDATE = 14
VARCHAR = 15
BIT = 16
TIMESTAMP2 = 17
DATETIME2 = 18
TIME2 = 19
JSON = 245
NEWDECIMAL = 246
ENUM = 247
SET = 248
TINY_BLOB = 249
MEDIUM_BLOB = 250
LONG_BLOB = 251
BLOB = 252
VAR_STRING = 253
STRING = 254
GEOMETRY = 255
venv/lib/python3.12/site-packages/mariadb/constants/INDICATOR.py
+19
View File
@@ -0,0 +1,19 @@
'''
MariaDB indicator variables
Indicator values are used in executemany() method of cursor class to
indicate special values.
'''
class MrdbIndicator():
indicator = 0
def __init__(self, indicator):
self.indicator = indicator
NULL = MrdbIndicator(1)
DEFAULT = MrdbIndicator(2)
IGNORE = MrdbIndicator(3)
IGNORE_ROW = MrdbIndicator(4)
venv/lib/python3.12/site-packages/mariadb/constants/INFO.py
+42
View File
@@ -0,0 +1,42 @@
"""
Constants for _get_info method of MariadB connection object
"""
CHARSET_ID = 0
CHARSET_NAME = 1
CLIENT_ERRORS = 2
CLIENT_VERSION = 3
CLIENT_VERSION_ID = 4
ASYNC_TIMEOUT = 5
ASYNC_TIMEOUT_MS = 6
CHARSET_INFO = 7
ERROR = 8
ERROR_ID = 9
HOST = 10
INFO = 11
PORT = 12
PROTOCOL_VERSION_ID = 13
PVIO_TYPE = 14
SCHEMA = 15
SERVER_TYPE = 16
SERVER_VERSION = 17
SERVER_VERSION_ID = 18
SOCKET = 19
SQLSTATE = 20
SSL_CIPHER = 21
TLS_LIBRARY = 22
TLS_VERSION = 23
TLS_VERSION_ID = 24
TYPE = 25
UNIX_SOCKET = 26
USER = 27
MAX_ALLOWED_PACKET = 28
NET_BUFFER_LENGTH = 29
SERVER_STATUS = 30
SERVER_CAPABILITIES = 31
EXTENDED_SERVER_CAPABILITIES = 32
CLIENT_CAPABILITIES = 33
BYTES_READ = 34
BYTES_SENT = 35
TLS_PEER_CERT_INFO = 36
TLS_VERIFY_STATUS = 37
venv/lib/python3.12/site-packages/mariadb/constants/STATUS.py
+21
View File
@@ -0,0 +1,21 @@
'''
MariaDB status flags
These flags describe the current status of the database server.
'''
IN_TRANS = 1
AUTOCOMMIT = 2
MORE_RESULTS_EXIST = 8
QUERY_NO_GOOD_INDEX_USED = 16
QUERY_NO_INDEX_USED = 32
CURSOR_EXISTS = 64
LAST_ROW_SENT = 128
DB_DROPPED = 256
NO_BACKSLASH_ESCAPES = 512
METADATA_CHANGED = 1024
QUERY_WAS_SLOW = 2048
PS_OUT_PARAMS = 4096
IN_TRANS_READONLY = 8192
SESSION_STATE_CHANGED = 16384
ANSI_QUOTES = 32768
venv/lib/python3.12/site-packages/mariadb/constants/TPC_STATE.py
+3
View File
@@ -0,0 +1,3 @@
NONE = 0
XID = 1
PREPARE = 2
venv/lib/python3.12/site-packages/mariadb/constants/__init__.py
+2
View File
@@ -0,0 +1,2 @@
__all__ = ["CLIENT", "CURSOR", "FIELD_TYPE", "FIELD_FLAG",
"INDICATOR", 'STATUS', 'ERR', 'CAPABILITY']
venv/lib/python3.12/site-packages/mariadb/cursors.py
+623
View File
@@ -0,0 +1,623 @@
#
# Copyright (C) 2020-2021 Georg Richter and MariaDB Corporation AB
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Library General Public
# License as published by the Free Software Foundation; either
# version 2 of the License, or (at your option) any later version.
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Library General Public License for more details.
# You should have received a copy of the GNU Library General Public
# License along with this library; if not see <http://www.gnu.org/licenses>
# or write to the Free Software Foundation, Inc.,
# 51 Franklin St., Fifth Floor, Boston, MA 02110, USA
#
import mariadb
import datetime
from numbers import Number
from mariadb.constants import CURSOR, STATUS, CAPABILITY, INDICATOR
from typing import Sequence
import decimal
PARAMSTYLE_QMARK = 1
PARAMSTYLE_FORMAT = 2
PARAMSTYLE_PYFORMAT = 3
ROWS_ALL = -1
RESULT_TUPLE = 0
RESULT_NAMEDTUPLE = 1
RESULT_DICTIONARY = 2
# Command types
SQL_NONE = 0,
SQL_INSERT = 1
SQL_UPDATE = 2
SQL_REPLACE = 3
SQL_DELETE = 4
SQL_CALL = 5
SQL_DO = 6
SQL_SELECT = 7
SQL_OTHER = 255
ROWS_EOF = -1
class Cursor(mariadb._mariadb.cursor):
"""
MariaDB Connector/Python Cursor Object
"""
def check_closed(self):
if self._thread_id != self.connection.thread_id:
raise mariadb.ProgrammingError(f"Cursor cannot be used anymore (the connection aborted and reconnected).")
if self._closed:
raise mariadb.ProgrammingError("Cursor cannot be used anymore (it was already closed before).")
self._connection._check_closed()
def __init__(self, connection, **kwargs):
"""
initialization
"""
self._bulk = False
self._dictionary = False
self._named_tuple = False
self._connection = connection
self._resulttype = RESULT_TUPLE
self._description = None
self._transformed_statement = None
self._prepared = False
self._prev_stmt = None
self._force_binary = None
self._rowcount = 0
self.buffered = True
self._parseinfo = None
self._data = None
self._thread_id = connection.thread_id
self._closed= None
if not connection:
raise mariadb.ProgrammingError("Invalid or no connection provided")
# parse keywords
if kwargs:
rtype = kwargs.pop("named_tuple", False)
if rtype:
self._resulttype = RESULT_NAMEDTUPLE
else:
rtype = kwargs.pop("dictionary", False)
if rtype:
self._resulttype = RESULT_DICTIONARY
buffered = kwargs.pop("buffered", True)
self.buffered = buffered
self._prepared = kwargs.pop("prepared", False)
self._force_binary = kwargs.pop("binary", False)
self._cursor_type = kwargs.pop("cursor_type", 0)
# call initialization of main class
super().__init__(connection, **kwargs)
def _substitute_parameters(self):
"""
Internal use only.
When running in text protocol, this method will replace placeholders
by supplied values.
For values which aren't numbers, strings or bytes string representation
will be used.
"""
new_stmt = self.statement.encode("utf8")
replace_diff = 0
if self._paramlist:
for i in range(0, len(self._paramlist)):
extra_bytes = 0
if self._paramstyle == PARAMSTYLE_PYFORMAT:
val = self._data[self._keys[i]]
else:
val = self._data[i]
if val is None:
replace = "NULL"
else:
if isinstance(val, INDICATOR.MrdbIndicator):
if val == INDICATOR.NULL:
replace = "NULL"
if val == INDICATOR.DEFAULT:
replace = "DEFAULT"
elif isinstance(val, Number):
replace = val.__str__()
else:
if isinstance(val, (bytes, bytearray)):
replace = "'%s'" % self.connection.escape_string(
val.decode(encoding='latin1'))
else:
replace = "'%s'" % self.connection.escape_string(
val.__str__())
extra_bytes = len(replace.encode("utf-8")) -\
len(replace)
ofs = self._paramlist[i] + replace_diff
new_stmt = new_stmt[:ofs] + replace.__str__().encode("utf8") +\
new_stmt[ofs+1:]
replace_diff += len(replace) - 1 + extra_bytes
return new_stmt
def _check_decimal_parameter(self, val):
"""
Internal use only
Checks for unsupported parameters.
The following parameters are not supported by MariaDB/MySQL (but also
not part of the SQL Standard)
- float("nan" | "inf" | "-inf")
- Decimal("NaN" | "sNaN" | "Infinity" | "-Infinity")
"""
if isinstance(val, float) and repr(val) in ("nan", "inf", "-inf"):
raise mariadb.NotSupportedError(f"Float value '{repr(val)}' is not supported.")
elif isinstance(val, decimal.Decimal) and val.__str__() in ("NaN", "sNaN", "Infinity", "-Infinity"):
raise mariadb.NotSupportedError(f"Decimal value '{val.__str__()}' is not supported.")
return None
def _check_execute_params(self):
# check data format
values = ()
if self._paramstyle in (PARAMSTYLE_QMARK, PARAMSTYLE_FORMAT):
if not isinstance(self._data, (tuple, list)):
raise mariadb.ProgrammingError("Data argument must be "
"Tuple or List")
values = self._data
if self._paramstyle == PARAMSTYLE_PYFORMAT:
if not isinstance(self._data, dict):
raise mariadb.ProgrammingError("Data argument must be "
"Dictionary")
for i in range(0, len(self._keys)):
if self._keys[i] not in self._data:
raise mariadb.ProgrammingError("Dictionary doesn't contain"
" key '%s'" % self._keys[i])
values = self._data.values()
else:
# check if number of place holders matches the number of
# supplied elements in data tuple
if self._paramlist and (
(not self._data and len(self._paramlist) > 0) or
(len(self._data) != len(self._paramlist))):
raise mariadb.ProgrammingError(
"statement (%s) doesn't match the number of data elements"
" (%s)." % (len(self._paramlist), len(self._data)))
# CONPY-313: Check if parameter types are supported
if len(values):
for val in values:
if isinstance(val, float) or isinstance(val, decimal.Decimal):
self._check_decimal_parameter(val)
def callproc(self, sp: str, data: Sequence = ()):
"""
Executes a stored procedure sp. The data sequence must contain an
entry for each parameter the procedure expects.
Input/Output or Output parameters have to be retrieved by .fetch
methods, the .sp_outparams attribute indicates if the result set
contains output parameters.
Arguments:
- sp: Name of stored procedure.
- data: Optional sequence containing data for placeholder
substitution.
"""
if self.connection.auto_reconnect:
self._thread_id= self.connection.thread_id
self.check_closed()
self._reset()
# create statement
params = ""
if data and len(data):
params = ("?," * len(data))[:-1]
statement = "CALL %s(%s)" % (sp, params)
self._rowcount = 0
self.execute(statement, data)
def _parse_execute(self, statement: str, data=(), is_bulk=False):
"""
For internal use
Parses SQL statement and checks parameters.
"""
if not statement:
raise mariadb.ProgrammingError("empty statement")
# parse statement
if self.statement != statement or is_bulk and not self._bulk:
super()._parse(statement)
self._prev_stmt = statement
self._reprepare = True
else:
self._reprepare = False
self._transformed_statement = self.statement
if self._cursor_type == CURSOR.READ_ONLY:
self._text = False
self._data = data
self._check_execute_params()
def nextset(self):
"""
Will make the cursor skip to the next available result set,
discarding any remaining rows from the current set.
"""
self.check_closed()
return super()._nextset()
def execute(self, statement: str, data: Sequence = (), buffered=None):
"""
Prepare and execute a SQL statement.
Parameters may be provided as sequence or mapping and will be bound
to variables in the operation. Variables are specified as question
marks (paramstyle ='qmark'), however for compatibility reasons MariaDB
Connector/Python also supports the 'format' and 'pyformat' paramstyles
with the restriction, that different paramstyles can't be mixed within
a statement.
A reference to the operation will be retained by the cursor.
If the cursor was created with attribute prepared =True the statement
string for following execute operations will be ignored.
This is most effective for algorithms where the same operation is used,
but different parameters are bound to it (many times).
By default execute() method generates an buffered result unless the
optional parameter buffered was set to False or the cursor was
generated as an unbuffered cursor.
"""
if self.connection.auto_reconnect:
self._thread_id= self.connection.thread_id
self.check_closed()
if not self._prepared:
self._reset()
self.connection._last_executed_statement = statement
# Parse statement
do_parse = True
self._rowcount = 0
if buffered is not None:
self.buffered = buffered
# clear pending result sets
if self.field_count:
self._clear_result()
# if we have a prepared cursor, we have to set statement
# to previous statement and don't need to parse
if self._prepared and self.statement:
statement = self.statement
do_parse = False
# parse statement and check param style
if do_parse:
self._parse_execute(statement, (data))
self._description = None
# CONPY-218: Allow None as replacement for empty tuple
data = data or ()
if len(data):
self._data = data
else:
self._data = None
# If statement doesn't contain parameters we force to run in text
# mode, unless a server side cursor or stored procedure will be
# executed.
if self._command != SQL_CALL and self._cursor_type == 0:
self._text = True
if self._force_binary:
self._text = False
# if one of the provided parameters has byte or datetime value,
# we don't use text protocol
if data and self._check_text_types() == True:
self._text = False
if self._text:
# in text mode we need to substitute parameters
# and store transformed statement
if (self.paramcount > 0):
self._transformed_statement = self._substitute_parameters()
else:
self._transformed_statement = self.statement
self._execute_text(self._transformed_statement)
self._readresponse()
else:
self._data = data
self._execute_binary()
self._initresult()
self._bulk = 0
def executemany(self, statement, parameters):
"""
Prepare a database operation (INSERT,UPDATE,REPLACE or DELETE
statement) and execute it against all parameter found in sequence.
Exactly behaves like .execute() but accepts a list of tuples, where
each tuple represents data of a row within a table.
.executemany() only supports DML (insert, update, delete) statements.
If the SQL statement contains a RETURNING clause, executemany()
returns a result set containing the values for columns listed in the
RETURNING clause.
"""
if self.connection.auto_reconnect:
self._thread_id= self.connection.thread_id
self.check_closed()
self._reset()
if not parameters or not len(parameters):
raise mariadb.ProgrammingError("No data provided")
self.connection._last_executed_statement = statement
# clear pending results
if self.field_count:
self._clear_result()
# If the server doesn't support bulk operations, we need to emulate
# by looping
# TODO: insert/replace statements are not optimized yet
# rowcount updating
if not (self.connection.extended_server_capabilities &
(CAPABILITY.BULK_OPERATIONS >> 32)):
count = 0
for row in parameters:
self.execute(statement, row)
count += self.rowcount
self._rowcount = count
else:
# parse statement
self._parse_execute(statement, parameters[0], is_bulk=True)
self._data = parameters
self.is_text = False
self._rowcount = 0
self._execute_bulk()
self._bulk = 1
def _fetch_row(self):
"""
Internal use only
fetches row and converts values, if connection has a converter.
"""
if not self.buffered:
self.check_closed()
# if there is no result set, PEP-249 requires to raise an
# exception
if not self.field_count:
raise mariadb.ProgrammingError("Cursor doesn't have a result set")
return super().fetchone()
def close(self):
"""
Closes the cursor.
If the cursor has pending or unread results, .close() will cancel them
so that further operations using the same connection can be executed.
The cursor will be unusable from this point forward; an Error
(or subclass) exception will be raised if any operation is attempted
with the cursor."
"""
# CONPY-231: fix memory leak
if self._data:
del self._data
if not self.connection._closed:
super().close()
self._closed= True
def fetchone(self):
"""
Fetch the next row of a query result set, returning a single sequence,
or None if no more data is available.
An exception will be raised if the previous call to execute() didn't
produce a result set or execute() wasn't called before.
"""
if not (self.buffered and self._text):
self.check_closed()
row = self._fetch_row()
return row
def fetchmany(self, size: int = 0):
"""
Fetch the next set of rows of a query result, returning a sequence
of sequences (e.g. a list of tuples). An empty sequence is returned
when no more rows are available.
The number of rows to fetch per call is specified by the parameter.
If it is not given, the cursor's arraysize determines the number
of rows to be fetched. The method should try to fetch as many rows
as indicated by the size parameter.
If this is not possible due to the specified number of rows not being
available, fewer rows may be returned.
An exception will be raised if the previous call to execute() didn't
produce a result set or execute() wasn't called before.
"""
if not (self.buffered and self._text):
self.check_closed()
if size == 0:
size = self.arraysize
return super().fetchrows(size)
def fetchall(self):
"""
Fetch all remaining rows of a query result, returning them as a
sequence of sequences (e.g. a list of tuples).
An exception will be raised if the previous call to execute() didn't
produce a result set or execute() wasn't called before.
"""
if not (self.buffered and self._text):
self.check_closed()
return super().fetchrows(ROWS_EOF)
def __iter__(self):
return iter(self.fetchone, None)
def scroll(self, value: int, mode="relative"):
"""
Scroll the cursor in the result set to a new position according to
mode.
If mode is "relative" (default), value is taken as offset to the
current position in the result set, if set to absolute, value states
an absolute target position.
"""
if not (self.buffered and self._text):
self.check_closed()
if self.field_count == 0:
raise mariadb.ProgrammingError("Cursor doesn't have a result set")
if not self.buffered:
raise mariadb.ProgrammingError("This method is available only "
"for cursors with a buffered "
"result set.")
if mode != "absolute" and mode != "relative":
raise mariadb.ProgrammingError("Invalid or unknown scroll "
"mode specified.")
if value == 0 and mode != "absolute":
raise mariadb.ProgrammingError("Invalid position value 0.")
if mode == "relative":
if self.rownumber + value < 0 or \
self.rownumber + value > self.rowcount:
raise mariadb.ProgrammingError("Position value "
"is out of range.")
new_pos = self.rownumber + value
else:
if value < 0 or value >= self.rowcount:
raise mariadb.ProgrammingError("Position value "
"is out of range.")
new_pos = value
self._seek(new_pos)
self._rownumber = new_pos
def setinputsizes(self, size: int):
"""
Required by PEP-249. Does nothing in MariaDB Connector/Python
"""
return
def setoutputsize(self, size: int):
"""
Required by PEP-249. Does nothing in MariaDB Connector/Python
"""
return
def __enter__(self):
"""Returns a copy of the cursor."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Closes cursor."""
self.close()
@property
def rowcount(self):
"""
This read-only attribute specifies the number of rows that the last\
execute*() produced (for DQL statements like SELECT) or affected
(for DML statements like UPDATE or INSERT).
The return value is -1 in case no .execute*() has been performed
on the cursor or the rowcount of the last operation cannot be
determined by the interface.
"""
# Even if PEP-249 permits operations on a closed cursor, we don't
# raise an exception if the cursor or the underlying connection
# was closed (See CONPY-269), instead we will return -1
if not self.buffered:
try:
self.check_closed()
except mariadb.ProgrammingError:
return -1
if self._rowcount > 0:
return self._rowcount
return super().rowcount
@property
def sp_outparams(self):
"""
Indicates if the current result set contains in out or out parameter
from a previous executed stored procedure
"""
self.check_closed()
return bool(self.connection.server_status & STATUS.PS_OUT_PARAMS)
@property
def lastrowid(self):
"""
Returns the ID generated by a query on a table with a column having
the AUTO_INCREMENT attribute or the value for the last usage of
LAST_INSERT_ID().
If the last query wasn't an INSERT or UPDATE
statement or if the modified table does not have a column with the
AUTO_INCREMENT attribute and LAST_INSERT_ID was not used, the returned
value will be None
"""
self.check_closed()
id = self.insert_id
if id > 0:
return id
return None
@property
def connection(self):
"""
Read-Only attribute which returns the reference to the connection
object on which the cursor was created.
"""
return self._connection
venv/lib/python3.12/site-packages/mariadb/dbapi20.py
+111
View File
@@ -0,0 +1,111 @@
from mariadb.constants import FIELD_TYPE
import time
import datetime
apilevel = '2.0'
paramstyle = 'qmark'
threadsafety = True
class DbApiType(frozenset):
"""
Immutable set for type checking
By default the following sets are defined:
- BINARY: for binary field types
- NUMBER: for numeric field types
- STRING: for character based (string) field types
- DATE: for date field type(s)
- DATETIME: for datetime and timestamp field type(s)
- TIME: for time field type(s)
- TIMESTAMP: for datetime and timestamp field type(s)
Example:
>>> FIELD_TYPE.GEOMETRY == mariadb.BINARY
True
>>> FIELD_TYPE.FLOAT == mariadb.BINARY
False
"""
def __eq__(self, field_type):
if (isinstance(field_type, DbApiType)):
return not self.difference(field_type)
return field_type in self
BINARY = DbApiType([FIELD_TYPE.GEOMETRY,
FIELD_TYPE.LONG_BLOB,
FIELD_TYPE.MEDIUM_BLOB,
FIELD_TYPE.TINY_BLOB,
FIELD_TYPE.BLOB])
STRING = DbApiType([FIELD_TYPE.ENUM,
FIELD_TYPE.JSON,
FIELD_TYPE.STRING,
FIELD_TYPE.VARCHAR,
FIELD_TYPE.VAR_STRING])
NUMBER = DbApiType([FIELD_TYPE.DECIMAL,
FIELD_TYPE.DOUBLE,
FIELD_TYPE.FLOAT,
FIELD_TYPE.INT24,
FIELD_TYPE.LONG,
FIELD_TYPE.LONGLONG,
FIELD_TYPE.NEWDECIMAL,
FIELD_TYPE.SHORT,
FIELD_TYPE.TINY,
FIELD_TYPE.YEAR])
DATE = DbApiType([FIELD_TYPE.DATE])
TIME = DbApiType([FIELD_TYPE.TIME])
DATETIME = TIMESTAMP = DbApiType([FIELD_TYPE.DATETIME,
FIELD_TYPE.TIMESTAMP])
ROWID = DbApiType()
def Binary(object):
"""Constructs an object capable of holding a binary value."""
return bytes(object)
def Date(year, month, day):
"""Constructs an object holding a date value."""
return datetime.date(year, month, day)
def Time(hour, minute, second):
"""Constructs an object holding a time value."""
return datetime.time(hour, minute, second)
def Timestamp(year, month, day, hour, minute, second):
"""Constructs an object holding a datetime value."""
return datetime.datetime(year, month, day, hour, minute, second)
def DateFromTicks(ticks):
"""Constructs an object holding a date value from the given ticks value
(number of seconds since the epoch).
For more information see the documentation of the standard Python
time module."""
return Date(*time.localtime(ticks)[:3])
def TimeFromTicks(ticks):
"""Constructs an object holding a time value from the given ticks value
(number of seconds since the epoch).
For more information see the documentation of the standard Python
time module."""
return Time(*time.localtime(ticks)[3:6])
def TimestampFromTicks(ticks):
"""Constructs an object holding a datetime value from the given ticks value
(number of seconds since the epoch).
For more information see the documentation of the standard Python
time module."""
return datetime.datetime(*time.localtime(ticks)[:6])
venv/lib/python3.12/site-packages/mariadb/field.py
+59
View File
@@ -0,0 +1,59 @@
from mariadb.constants import FIELD_TYPE, FIELD_FLAG
field_types = {FIELD_TYPE.DECIMAL: "DECIMAL",
FIELD_TYPE.TINY: "TINY",
FIELD_TYPE.SHORT: "SHORT",
FIELD_TYPE.LONG: "LONG",
FIELD_TYPE.FLOAT: "FLOAT",
FIELD_TYPE.DOUBLE: "DOUBLE",
FIELD_TYPE.NULL: "NULL",
FIELD_TYPE.TIMESTAMP: "TIMESTAMP",
FIELD_TYPE.LONGLONG: "LONGLONG",
FIELD_TYPE.INT24: "INT24",
FIELD_TYPE.DATE: "DATE",
FIELD_TYPE.TIME: "TIME",
FIELD_TYPE.DATETIME: "DATETIME",
FIELD_TYPE.YEAR: "YEAR",
FIELD_TYPE.NEWDATE: "NEWDATE",
FIELD_TYPE.VARCHAR: "VARCHAR",
FIELD_TYPE.BIT: "BIT",
FIELD_TYPE.JSON: "JSON",
FIELD_TYPE.NEWDECIMAL: "NEWDECIMAL",
FIELD_TYPE.ENUM: "ENUM",
FIELD_TYPE.SET: "SET",
FIELD_TYPE.TINY_BLOB: "TINY_BLOB",
FIELD_TYPE.MEDIUM_BLOB: "MEDIUM_BLOB",
FIELD_TYPE.LONG_BLOB: "LONG_BLOB",
FIELD_TYPE.BLOB: "BLOB",
FIELD_TYPE.VAR_STRING: "VAR_STRING",
FIELD_TYPE.STRING: "STRING",
FIELD_TYPE.GEOMETRY: "GEOMETRY"}
field_flags = {FIELD_FLAG.NOT_NULL: "NOT_NULL",
FIELD_FLAG.PRIMARY_KEY: "PRIMARY_KEY",
FIELD_FLAG.UNIQUE_KEY: "UNIQUE_KEY",
FIELD_FLAG.MULTIPLE_KEY: "PART_KEY",
FIELD_FLAG.BLOB: "BLOB",
FIELD_FLAG.UNSIGNED: "UNSIGNED",
FIELD_FLAG.ZEROFILL: "ZEROFILL",
FIELD_FLAG.BINARY: "BINARY",
FIELD_FLAG.ENUM: "NUMERIC",
FIELD_FLAG.AUTO_INCREMENT: "AUTO_INCREMENT",
FIELD_FLAG.TIMESTAMP: "TIMESTAMP",
FIELD_FLAG.SET: "SET",
FIELD_FLAG.NO_DEFAULT: "NO_DEFAULT",
FIELD_FLAG.ON_UPDATE_NOW: "UPDATE_TIMESTAMP",
FIELD_FLAG.NUMERIC: "NUMERIC"}
class fieldinfo():
def type(self, description):
if description[1] in field_types:
return field_types[description[1]]
return None
def flag(self, description):
flags = [field_flags[f] for f in field_flags.keys()
if description[7] & f]
return " | ".join(flags)
venv/lib/python3.12/site-packages/mariadb/release_info.py
+3
View File
@@ -0,0 +1,3 @@
__author__ = 'Georg Richter'
__version__ = '1.1.14'
__version_info__ = (1, 1, 14)