tessl/pypi-dataset
Toolkit for Python-based database access that makes reading and writing data in databases as simple as working with JSON files
—
Pending
database-operations.mddocs/
Database Operations
Database connection management, table access, transaction control, and raw SQL query execution. The Database class provides the foundation for all dataset operations with automatic schema management and connection pooling.
Capabilities
Connection Management
Create and manage database connections with support for multiple database backends and connection pooling.
def connect(url=None, schema=None, engine_kwargs=None, ensure_schema=True,
row_type=OrderedDict, sqlite_wal_mode=True, on_connect_statements=None):
"""
Opens a new connection to a database.
Parameters:
- url: str, SQLAlchemy engine URL (defaults to DATABASE_URL env var or sqlite://)
- schema: str, database schema name
- engine_kwargs: dict, additional SQLAlchemy engine parameters
- ensure_schema: bool, automatically create tables and columns (default True)
- row_type: type, dict-like class for row containers (default OrderedDict)
- sqlite_wal_mode: bool, enable WAL mode for SQLite (default True)
- on_connect_statements: list, SQL statements to run on connect
Returns:
Database instance
"""
class Database:
def __init__(self, url, schema=None, engine_kwargs=None, ensure_schema=True,
row_type=OrderedDict, sqlite_wal_mode=True, on_connect_statements=None):
"""Configure and connect to the database."""
def close(self):
"""Close database connections. Makes this object unusable."""Table Access
Access and manage database tables with automatic creation and schema reflection.
class Database:
def has_table(self, name):
"""Check if a table exists in the database."""
def create_table(self, table_name, primary_id=None, primary_type=None, primary_increment=None):
"""
Create a new table.
Parameters:
- table_name: str, name of the table
- primary_id: str, name of primary key column (default "id")
- primary_type: SQLAlchemy type, type of primary key (default Integer)
- primary_increment: bool, auto-increment primary key (default True for numeric types)
Returns:
Table instance
"""
def load_table(self, table_name):
"""
Load an existing table.
Parameters:
- table_name: str, name of existing table
Returns:
Table instance
"""
def get_table(self, table_name, primary_id=None, primary_type=None, primary_increment=None):
"""
Load or create a table (same as create_table).
Returns:
Table instance
"""
def __getitem__(self, table_name):
"""Get a table using bracket notation: db['table_name']"""
def __contains__(self, table_name):
"""Check if table exists using 'in' operator: 'table' in db"""Database Properties
Access database metadata and connection information.
class Database:
@property
def tables(self):
"""Get a list of all table names in the database."""
@property
def views(self):
"""Get a list of all view names in the database."""
@property
def executable(self):
"""Connection against which statements will be executed."""
@property
def metadata(self):
"""Return a SQLAlchemy schema cache object."""
@property
def in_transaction(self):
"""Check if this database is in a transactional context."""
@property
def op(self):
"""Get an alembic operations context."""
@property
def inspect(self):
"""Get a SQLAlchemy inspector."""
@property
def types(self):
"""Get the Types instance for accessing SQLAlchemy types."""Transaction Management
Handle database transactions with explicit control or context manager support.
class Database:
def begin(self):
"""
Enter a transaction explicitly.
No data will be written until the transaction has been committed.
"""
def commit(self):
"""
Commit the current transaction.
Make all statements executed since the transaction was begun permanent.
"""
def rollback(self):
"""
Roll back the current transaction.
Discard all statements executed since the transaction was begun.
"""
def __enter__(self):
"""Start a transaction."""
def __exit__(self, error_type, error_value, traceback):
"""End a transaction by committing or rolling back."""Raw SQL Queries
Execute arbitrary SQL statements with parameter binding and result iteration.
class Database:
def query(self, query, *args, **kwargs):
"""
Run a statement on the database directly.
Parameters:
- query: str or SQLAlchemy expression, SQL query to execute
- *args: positional parameters for query binding
- **kwargs: keyword parameters for query binding
- _step: int, result fetching step size (default 1000)
Returns:
ResultIter: Iterator yielding each result row as dict-like objects
"""Usage Examples
Database Connection
import dataset
# SQLite (default)
db = dataset.connect() # In-memory
db = dataset.connect('sqlite:///data.db') # File-based
# PostgreSQL
db = dataset.connect('postgresql://user:pass@localhost/dbname')
# MySQL
db = dataset.connect('mysql://user:pass@localhost/dbname')
# With custom options
db = dataset.connect(
'sqlite:///data.db',
engine_kwargs={'pool_recycle': 3600},
row_type=dict, # Use regular dict instead of OrderedDict
on_connect_statements=['PRAGMA foreign_keys=ON']
)Table Management
# Check if table exists
if 'users' in db:
print("Users table exists")
# Create table with custom primary key
users = db.create_table('users', 'user_id', db.types.string(50))
# Load existing table
existing_table = db.load_table('products')
# Get or create table (most common)
orders = db['orders'] # Shorthand syntax
orders = db.get_table('orders') # Explicit syntaxTransactions
# Explicit transaction control
db.begin()
try:
table.insert({'name': 'John'})
table.insert({'name': 'Jane'})
db.commit()
except Exception:
db.rollback()
raise
# Context manager (recommended)
with db:
table.insert({'name': 'John'})
table.insert({'name': 'Jane'})
# Automatically commits on success, rolls back on exceptionRaw SQL Queries
# Simple query
for row in db.query('SELECT * FROM users WHERE age > 18'):
print(row['name'])
# Parameterized queries
for row in db.query('SELECT * FROM users WHERE city = ?', 'New York'):
print(row)
# Named parameters
for row in db.query('SELECT * FROM users WHERE age > :min_age', min_age=21):
print(row)
# Complex query with SQLAlchemy expression
from sqlalchemy import text
query = text('SELECT COUNT(*) as total FROM users WHERE created > :date')
result = list(db.query(query, date='2023-01-01'))
print(result[0]['total'])Install with Tessl CLI
npx tessl i tessl/pypi-dataset