tessl/pypi-mysql-connector
MySQL driver written in Python providing comprehensive database connectivity with both traditional SQL and modern document operations
—
Pending
document-operations.mddocs/
Document Operations (X DevAPI)
Modern document database operations using MySQL's X DevAPI, supporting JSON document storage, retrieval, and manipulation with a fluent API design.
Capabilities
Session Management
Create and manage X Protocol sessions for document and relational operations.
def get_session(*args, **kwargs):
"""
Create an X Protocol session with load balancing support.
Parameters:
- *args: Connection string or configuration dict
- **kwargs: Connection parameters (host, port, user, password, schema)
Returns:
XSession: X Protocol session with multi-host support
Raises:
InterfaceError: Invalid connection parameters
"""
def get_node_session(*args, **kwargs):
"""
Create a single-node X Protocol session.
Parameters:
- *args: Connection string or configuration dict
- **kwargs: Connection parameters (host, port, user, password, schema)
Returns:
NodeSession: Single-node X Protocol session
Raises:
InterfaceError: Invalid connection parameters or multiple hosts specified
"""
class XSession:
"""
X Protocol session with load balancing and high availability support.
Provides document and relational data operations through MySQL X Protocol,
supporting automatic failover across multiple MySQL server instances.
"""
def get_schema(self, name):
"""
Get a schema object.
Parameters:
- name (str): Schema name
Returns:
Schema: Schema object for database operations
"""
def get_schemas(self):
"""
Get list of available schemas.
Returns:
list[Schema]: List of schema objects
"""
def create_schema(self, name):
"""
Create a new schema.
Parameters:
- name (str): Schema name to create
Returns:
Schema: Created schema object
"""
def drop_schema(self, name):
"""Drop a schema"""
def sql(self, statement):
"""
Execute raw SQL statement.
Parameters:
- statement (str): SQL statement to execute
Returns:
SqlStatement: SQL statement object for execution
"""
def close(self):
"""Close the session and free resources"""
class NodeSession(XSession):
"""Single-node X Protocol session without load balancing"""Usage Example:
import mysqlx
# Create session with connection string
session = mysqlx.get_session('mysqlx://user:password@localhost:33060/mydb')
# Create session with parameters
session = mysqlx.get_session({
'host': 'localhost',
'port': 33060,
'user': 'myuser',
'password': 'mypassword',
'schema': 'mydb'
})
# Get schema
schema = session.get_schema('mydb')
# Close session
session.close()Schema Operations
Database schema management and object access.
class Schema:
"""
Database schema representation for X DevAPI operations.
Provides access to collections, tables, and views within a schema,
supporting both document and relational data operations.
"""
def get_collections(self):
"""
Get list of collections in schema.
Returns:
list[Collection]: List of collection objects
"""
def get_collection(self, name):
"""
Get a collection object.
Parameters:
- name (str): Collection name
Returns:
Collection: Collection object for document operations
"""
def create_collection(self, name, reuse_existing=False):
"""
Create a new collection.
Parameters:
- name (str): Collection name
- reuse_existing (bool): Return existing collection if exists
Returns:
Collection: Created or existing collection object
"""
def drop_collection(self, name):
"""Drop a collection"""
def get_tables(self):
"""
Get list of tables in schema.
Returns:
list[Table]: List of table objects
"""
def get_table(self, name):
"""
Get a table object.
Parameters:
- name (str): Table name
Returns:
Table: Table object for relational operations
"""
def get_view(self, name):
"""
Get a view object.
Parameters:
- name (str): View name
Returns:
View: View object for read operations
"""
@property
def name(self):
"""Schema name"""
def exists_in_database(self):
"""Check if schema exists in database"""Collection Operations
Document collection management and CRUD operations.
class Collection:
"""
Document collection for storing and querying JSON documents.
Provides NoSQL-style operations for document storage, retrieval,
and manipulation using MySQL's document store capabilities.
"""
def add(self, *docs):
"""
Add documents to collection.
Parameters:
- *docs: Documents to add (dict, DbDoc, or JSON strings)
Returns:
AddStatement: Statement for execution
"""
def find(self, condition=None):
"""
Find documents in collection.
Parameters:
- condition (str): Search condition expression
Returns:
FindStatement: Statement for further refinement and execution
"""
def modify(self, condition):
"""
Modify documents in collection.
Parameters:
- condition (str): Condition for documents to modify
Returns:
ModifyStatement: Statement for specifying modifications
"""
def remove(self, condition):
"""
Remove documents from collection.
Parameters:
- condition (str): Condition for documents to remove
Returns:
RemoveStatement: Statement for execution
"""
def create_index(self, name, fields):
"""
Create an index on collection.
Parameters:
- name (str): Index name
- fields (dict): Index field specification
Returns:
CreateCollectionIndexStatement: Statement for execution
"""
def drop_index(self, name):
"""
Drop an index from collection.
Parameters:
- name (str): Index name to drop
Returns:
DropCollectionIndexStatement: Statement for execution
"""
def count(self):
"""
Count documents in collection.
Returns:
int: Number of documents
"""
def exists_in_database(self):
"""Check if collection exists in database"""
@property
def name(self):
"""Collection name"""
@property
def schema(self):
"""Parent schema object"""Usage Example:
# Get collection
collection = schema.get_collection('users')
# Add documents
collection.add({
'name': 'John Doe',
'email': 'john@example.com',
'age': 30,
'address': {
'city': 'New York',
'state': 'NY'
}
}).execute()
# Find documents
docs = collection.find('age > 25').execute()
for doc in docs:
print(f"Name: {doc['name']}, Age: {doc['age']}")
# Modify documents
collection.modify('age < 30').set('status', 'young').execute()
# Remove documents
collection.remove('age > 65').execute()Document Query Operations
Fluent API for building complex document queries.
class FindStatement:
"""
Fluent interface for finding documents in collections.
Provides chainable methods for building complex queries with filtering,
sorting, limiting, and field projection.
"""
def fields(self, *fields):
"""
Select specific fields to return.
Parameters:
- *fields: Field names or expressions to select
Returns:
FindStatement: Self for method chaining
"""
def group_by(self, *fields):
"""
Group results by fields.
Parameters:
- *fields: Fields to group by
Returns:
FindStatement: Self for method chaining
"""
def having(self, condition):
"""
Filter grouped results.
Parameters:
- condition (str): HAVING condition expression
Returns:
FindStatement: Self for method chaining
"""
def sort(self, *sort_criteria):
"""
Sort results.
Parameters:
- *sort_criteria: Sort expressions ("field ASC", "field DESC")
Returns:
FindStatement: Self for method chaining
"""
def limit(self, count, offset=0):
"""
Limit number of results.
Parameters:
- count (int): Maximum number of documents
- offset (int): Number of documents to skip
Returns:
FindStatement: Self for method chaining
"""
def bind(self, *args, **kwargs):
"""
Bind values to placeholders.
Parameters:
- *args: Positional parameter values
- **kwargs: Named parameter values
Returns:
FindStatement: Self for method chaining
"""
def execute(self):
"""
Execute the find operation.
Returns:
DocResult: Result object with documents
"""
class AddStatement:
"""Statement for adding documents to collection"""
def execute(self):
"""
Execute the add operation.
Returns:
Result: Operation result with generated IDs
"""
class ModifyStatement:
"""Statement for modifying documents in collection"""
def set(self, field, value):
"""Set field to value"""
def unset(self, *fields):
"""Remove fields from documents"""
def array_insert(self, field, value):
"""Insert value into array field"""
def array_append(self, field, value):
"""Append value to array field"""
def array_delete(self, field):
"""Delete array element"""
def patch(self, patch_expr):
"""Apply JSON patch to documents"""
def sort(self, *sort_criteria):
"""Sort documents before modification"""
def limit(self, count):
"""Limit number of documents to modify"""
def bind(self, *args, **kwargs):
"""Bind parameter values"""
def execute(self):
"""Execute the modify operation"""
class RemoveStatement:
"""Statement for removing documents from collection"""
def sort(self, *sort_criteria):
"""Sort documents before removal"""
def limit(self, count):
"""Limit number of documents to remove"""
def bind(self, *args, **kwargs):
"""Bind parameter values"""
def execute(self):
"""Execute the remove operation"""Table Operations (Relational)
Relational table operations through X DevAPI.
class Table:
"""
Relational table for SQL-like operations through X DevAPI.
Provides CRUD operations for relational data using the same
fluent interface as document collections.
"""
def select(self, *fields):
"""
Select data from table.
Parameters:
- *fields: Column names or expressions to select
Returns:
SelectStatement: Statement for building SELECT query
"""
def insert(self, *fields):
"""
Insert data into table.
Parameters:
- *fields: Column names for insertion
Returns:
InsertStatement: Statement for specifying values
"""
def update(self):
"""
Update data in table.
Returns:
UpdateStatement: Statement for specifying updates
"""
def delete(self):
"""
Delete data from table.
Returns:
DeleteStatement: Statement for specifying conditions
"""
def count(self):
"""Count rows in table"""
def exists_in_database(self):
"""Check if table exists in database"""
@property
def name(self):
"""Table name"""
class SelectStatement:
"""Statement for selecting data from tables"""
def where(self, condition):
"""Filter rows with WHERE condition"""
def group_by(self, *fields):
"""Group results by columns"""
def having(self, condition):
"""Filter grouped results"""
def order_by(self, *sort_criteria):
"""Sort results"""
def limit(self, count, offset=0):
"""Limit number of results"""
def bind(self, *args, **kwargs):
"""Bind parameter values"""
def execute(self):
"""Execute SELECT statement"""
class InsertStatement:
"""Statement for inserting data into tables"""
def values(self, *values):
"""Specify values to insert"""
def execute(self):
"""Execute INSERT statement"""
class UpdateStatement:
"""Statement for updating table data"""
def set(self, field, value):
"""Set column to value"""
def where(self, condition):
"""Filter rows to update"""
def order_by(self, *sort_criteria):
"""Sort rows before update"""
def limit(self, count):
"""Limit number of rows to update"""
def bind(self, *args, **kwargs):
"""Bind parameter values"""
def execute(self):
"""Execute UPDATE statement"""
class DeleteStatement:
"""Statement for deleting table data"""
def where(self, condition):
"""Filter rows to delete"""
def order_by(self, *sort_criteria):
"""Sort rows before deletion"""
def limit(self, count):
"""Limit number of rows to delete"""
def bind(self, *args, **kwargs):
"""Bind parameter values"""
def execute(self):
"""Execute DELETE statement"""Usage Example:
# Get table
table = schema.get_table('users')
# Select data
result = table.select('id', 'name', 'email').where('age > :age').bind(age=25).execute()
for row in result:
print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}")
# Insert data
table.insert('name', 'email', 'age').values('John Doe', 'john@example.com', 30).execute()
# Update data
table.update().set('status', 'active').where('age > 18').execute()
# Delete data
table.delete().where('age < 13').execute()Result Handling
Result objects returned from X DevAPI operations, providing access to data, metadata, and execution information.
class Result:
"""
Base result object for X DevAPI operations.
Provides common functionality for accessing operation results including
affected rows, generated identifiers, and warning information.
"""
def get_affected_items_count(self) -> int:
"""Get number of items affected by operation"""
def get_autoincrement_value(self) -> int:
"""Get auto-generated ID from operation"""
def get_generated_ids(self) -> list:
"""Get list of generated document IDs"""
def get_warnings(self) -> list:
"""Get list of warnings from operation"""
def get_warnings_count(self) -> int:
"""Get count of warnings generated"""
class RowResult(Result):
"""
Result containing tabular row data from relational operations.
Provides access to rows, columns, and metadata from SELECT
operations on tables and views.
"""
def fetch_one(self) -> Row:
"""Fetch next row from result set"""
def fetch_all(self) -> list[Row]:
"""Fetch all remaining rows from result set"""
def get_columns(self) -> list[ColumnMetaData]:
"""Get column metadata for result set"""
def get_column_names(self) -> list[str]:
"""Get list of column names"""
def get_column_count(self) -> int:
"""Get number of columns in result set"""
def count(self) -> int:
"""Get total number of rows in result set"""
class SqlResult(RowResult):
"""
Result from SQL statement execution.
Extends RowResult with SQL-specific functionality and
provides access to multiple result sets from stored procedures.
"""
def next_result(self) -> bool:
"""Move to next result set if available"""
def has_data(self) -> bool:
"""Check if result contains data rows"""
class DocResult(Result):
"""
Result containing document data from collection operations.
Provides access to JSON documents returned from collection
find operations and document modifications.
"""
def fetch_one(self) -> DbDoc:
"""Fetch next document from result set"""
def fetch_all(self) -> list[DbDoc]:
"""Fetch all remaining documents from result set"""
def count(self) -> int:
"""Get total number of documents in result set"""
class Row:
"""
Row of tabular data from relational operations.
Provides access to column values by index or name,
with automatic type conversion from MySQL data types.
"""
def get_string(self, index: int) -> str:
"""Get column value as string"""
def get_int(self, index: int) -> int:
"""Get column value as integer"""
def get_float(self, index: int) -> float:
"""Get column value as float"""
def __getitem__(self, key):
"""Get column value by index or name"""
class ColumnMetaData:
"""
Metadata information for result set columns.
Provides column properties including name, type,
length, precision, and other attributes.
"""
def get_column_name(self) -> str:
"""Get column name"""
def get_column_label(self) -> str:
"""Get column label/alias"""
def get_type(self) -> ColumnType:
"""Get column data type"""
def get_length(self) -> int:
"""Get column length"""
def is_number_signed(self) -> bool:
"""Check if numeric column is signed"""
def get_fractional_digits(self) -> int:
"""Get number of fractional digits for decimal types"""
class ColumnType:
"""Column data type constants for result metadata"""
BIT: int
TINYINT: int
SMALLINT: int
MEDIUMINT: int
INT: int
BIGINT: int
FLOAT: int
DOUBLE: int
DECIMAL: int
DATE: int
TIME: int
DATETIME: int
TIMESTAMP: int
CHAR: int
VARCHAR: int
BINARY: int
VARBINARY: int
BLOB: int
TEXT: int
ENUM: int
SET: int
JSON: int
GEOMETRY: intTypes
SessionConfig = {
'host': str,
'port': int,
'user': str,
'password': str,
'schema': str,
'ssl_mode': str,
'ssl_ca': str,
'ssl_cert': str,
'ssl_key': str,
'connect_timeout': int,
'compression': str,
'auth': str
}
class DbDoc(dict):
"""
JSON document representation extending dict.
Provides dictionary interface with additional methods for
document manipulation and MySQL-specific functionality.
"""
def copy(self):
"""Create a copy of the document"""
def ensure_id(self):
"""Ensure document has _id field"""Install with Tessl CLI
npx tessl i tessl/pypi-mysql-connector