Elysia Tools
Navegação
Python
Exemplos de Banco de Dados Web Python
Exemplos de banco de dados Web Python incluindo conexão SQLite, consultas SQL e manipulação de transações
Exemplos
Entradas desta coleção
CódigopythonSimples
Conexão SQLite
Conectar ao banco de dados SQLite, criar tabelas e gerenciar conexões de banco de dados com tratamento de erros adequado
Dificuldade
3/10
Tempo estimado
20 min
Etiquetas
python, web, database, sqlite
Pré-requisitos
Basic Python, SQL basics
python
# Web Python SQLite Connection Examples
# Connecting to SQLite database and managing database connections
# 1. Basic Connection
import sqlite3
from typing import Optional, List, Tuple, Any
from contextlib import contextmanager
import os
def connect_to_database(db_path: str = ':memory:') -> sqlite3.Connection:
"""
Connect to SQLite database
Args:
db_path: Database file path (:memory: for in-memory)
Returns:
Connection object
"""
connection = sqlite3.connect(db_path)
return connection
def close_connection(connection: sqlite3.Connection) -> None:
"""
Close database connection
Args:
connection: Connection to close
"""
if connection:
connection.close()
def is_connection_open(connection: sqlite3.Connection) -> bool:
"""
Check if connection is open
Args:
connection: Connection to check
Returns:
True if open
"""
try:
connection.execute('SELECT 1')
return True
except:
return False
# 2. Connection with Context Manager
@contextmanager
def database_connection(db_path: str = ':memory:'):
"""
Context manager for database connection
Args:
db_path: Database file path
Yields:
Connection object
"""
conn = None
try:
conn = sqlite3.connect(db_path)
yield conn
finally:
if conn:
conn.close()
def with_connection(func):
"""
Decorator for database connection
Args:
func: Function to wrap
Returns:
Wrapped function
"""
def wrapper(*args, **kwargs):
db_path = kwargs.get('db_path', ':memory:')
with database_connection(db_path) as conn:
kwargs['conn'] = conn
return func(*args, **kwargs)
return wrapper
# 3. Connection Configuration
def configure_connection(conn: sqlite3.Connection, **settings) -> None:
"""
Configure connection settings
Args:
conn: Connection object
**settings: Configuration settings
"""
# Enable foreign keys
if settings.get('foreign_keys', True):
conn.execute('PRAGMA foreign_keys = ON')
# Set row factory
if settings.get('row_factory'):
conn.row_factory = settings['row_factory']
# Set timeout (milliseconds)
if 'timeout' in settings:
conn.execute("PRAGMA busy_timeout = ?", (int(settings['timeout'] * 1000),))
# Set isolation level
if 'isolation_level' in settings:
conn.isolation_level = settings['isolation_level']
def get_connection_info(conn: sqlite3.Connection) -> dict:
"""
Get connection information
Args:
conn: Connection object
Returns:
Connection info dictionary
"""
cursor = conn.cursor()
cursor.execute('PRAGMA database_list')
databases = cursor.fetchall()
cursor.execute('SELECT sqlite_version()')
version = cursor.fetchone()[0]
return {
'databases': databases,
'sqlite_version': version,
'isolation_level': conn.isolation_level,
'total_changes': conn.total_changes,
}
# 4. Table Operations
def create_table(conn: sqlite3.Connection, table_name: str, schema: str) -> bool:
"""
Create table
Args:
conn: Connection object
table_name: Table name
schema: Table schema (SQL)
Returns:
True if successful
"""
try:
cursor = conn.cursor()
cursor.execute(f'CREATE TABLE IF NOT EXISTS {table_name} ({schema})')
conn.commit()
return True
except Exception as e:
print(f"Error creating table: {e}")
return False
def drop_table(conn: sqlite3.Connection, table_name: str) -> bool:
"""
Drop table
Args:
conn: Connection object
table_name: Table name
Returns:
True if successful
"""
try:
cursor = conn.cursor()
cursor.execute(f'DROP TABLE IF EXISTS {table_name}')
conn.commit()
return True
except Exception as e:
print(f"Error dropping table: {e}")
return False
def table_exists(conn: sqlite3.Connection, table_name: str) -> bool:
"""
Check if table exists
Args:
conn: Connection object
table_name: Table name
Returns:
True if exists
"""
cursor = conn.cursor()
cursor.execute("""
SELECT name FROM sqlite_master
WHERE type='table' AND name=?
""", (table_name,))
return cursor.fetchone() is not None
def get_table_schema(conn: sqlite3.Connection, table_name: str) -> List[Tuple]:
"""
Get table schema
Args:
conn: Connection object
table_name: Table name
Returns:
List of column info
"""
cursor = conn.cursor()
cursor.execute(f'PRAGMA table_info({table_name})')
return cursor.fetchall()
def list_tables(conn: sqlite3.Connection) -> List[str]:
"""
List all tables
Args:
conn: Connection object
Returns:
List of table names
"""
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
return [row[0] for row in cursor.fetchall()]
# 5. Database File Operations
def create_database_file(file_path: str) -> bool:
"""
Create new database file
Args:
file_path: Path to database file
Returns:
True if successful
"""
try:
conn = sqlite3.connect(file_path)
conn.close()
return True
except Exception as e:
print(f"Error creating database: {e}")
return False
def delete_database_file(file_path: str) -> bool:
"""
Delete database file
Args:
file_path: Path to database file
Returns:
True if deleted
"""
try:
if os.path.exists(file_path):
os.remove(file_path)
return True
return False
except Exception as e:
print(f"Error deleting database: {e}")
return False
def database_file_exists(file_path: str) -> bool:
"""
Check if database file exists
Args:
file_path: Path to check
Returns:
True if exists
"""
return os.path.exists(file_path) and file_path.endswith('.db')
def get_database_size(file_path: str) -> int:
"""
Get database file size
Args:
file_path: Path to database
Returns:
Size in bytes
"""
if os.path.exists(file_path):
return os.path.getsize(file_path)
return 0
# 6. Row Factory Configuration
def set_dict_row_factory(conn: sqlite3.Connection) -> None:
"""
Set row factory to return dictionaries
Args:
conn: Connection object
"""
conn.row_factory = sqlite3.Row
def set_tuple_row_factory(conn: sqlite3.Connection) -> None:
"""
Set row factory to return tuples (default)
Args:
conn: Connection object
"""
conn.row_factory = None
# 7. Connection Pooling (Simple Implementation)
class ConnectionPool:
"""Simple connection pool"""
def __init__(self, db_path: str, pool_size: int = 5):
"""
Initialize connection pool
Args:
db_path: Database path
pool_size: Maximum connections
"""
self.db_path = db_path
self.pool_size = pool_size
self.pool: List[sqlite3.Connection] = []
self.used: List[sqlite3.Connection] = []
def get_connection(self) -> sqlite3.Connection:
"""Get connection from pool"""
if self.pool:
conn = self.pool.pop()
self.used.append(conn)
return conn
if len(self.used) < self.pool_size:
conn = sqlite3.connect(self.db_path)
self.used.append(conn)
return conn
raise Exception("Connection pool exhausted")
def return_connection(self, conn: sqlite3.Connection) -> None:
"""Return connection to pool"""
if conn in self.used:
self.used.remove(conn)
self.pool.append(conn)
def close_all(self) -> None:
"""Close all connections"""
for conn in self.pool + self.used:
conn.close()
self.pool.clear()
self.used.clear()
# 8. Backup and Restore
def backup_database(source: str, backup: str) -> bool:
"""
Create database backup
Args:
source: Source database path
backup: Backup database path
Returns:
True if successful
"""
try:
source_conn = sqlite3.connect(source)
backup_conn = sqlite3.connect(backup)
source_conn.backup(backup_conn)
backup_conn.close()
source_conn.close()
return True
except Exception as e:
print(f"Backup error: {e}")
return False
def restore_database(backup: str, target: str) -> bool:
"""
Restore database from backup
Args:
backup: Backup database path
target: Target database path
Returns:
True if successful
"""
return backup_database(backup, target)
# Usage Examples
def demonstrate_sqlite_connect():
print("=== Web Python SQLite Connection Examples ===\n")
# 1. Basic connection
print("--- 1. Basic Connection ---")
conn = connect_to_database()
print(f"Connected: {is_connection_open(conn)}")
print(f"Connection info: {get_connection_info(conn)}")
close_connection(conn)
# 2. Context manager
print("\n--- 2. Context Manager ---")
with database_connection() as conn:
print(f"Connected in context: {is_connection_open(conn)}")
print(f"Closed after context: {not is_connection_open(conn)}")
# 3. Configure connection
print("\n--- 3. Configure Connection ---")
with database_connection() as conn:
configure_connection(conn, row_factory=sqlite3.Row)
print(f"Configured connection: {get_connection_info(conn)}")
# 4. Table operations
print("\n--- 4. Table Operations ---")
with database_connection() as conn:
# Create table
create_table(conn, 'users', '''
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE
''')
print(f"Table exists: {table_exists(conn, 'users')}")
print(f"Table schema: {get_table_schema(conn, 'users')}")
print(f"All tables: {list_tables(conn)}")
# 5. Row factory
print("\n--- 5. Row Factory ---")
with database_connection() as conn:
create_table(conn, 'test', 'id INTEGER, value TEXT')
conn.execute('INSERT INTO test VALUES (1, "test")')
conn.commit()
# Default (tuple)
cursor = conn.execute('SELECT * FROM test')
print(f"Default row factory: {cursor.fetchone()}")
# Dict row factory
set_dict_row_factory(conn)
cursor = conn.execute('SELECT * FROM test')
row = cursor.fetchone()
print(f"Dict row factory: {dict(row)}")
# 6. Connection pool
print("\n--- 6. Connection Pool ---")
pool = ConnectionPool(':memory:', pool_size=2)
conn1 = pool.get_connection()
conn2 = pool.get_connection()
print(f"Pool size: {len(pool.used)}")
pool.return_connection(conn1)
print(f"After return: {len(pool.used)}")
pool.close_all()
print("\n=== All SQLite Connection Examples Completed ===")
# Export functions
# export { connect_to_database, close_connection, is_connection_open }
# export { database_connection, with_connection }
# export { configure_connection, get_connection_info }
# export { create_table, drop_table, table_exists, get_table_schema, list_tables }
# export { create_database_file, delete_database_file, database_file_exists, get_database_size }
# export { set_dict_row_factory, set_tuple_row_factory }
# export { ConnectionPool }
# export { backup_database, restore_database }
# export { demonstrate_sqlite_connect }
CódigopythonIntermediário
Executar Consulta SQL
Executar consultas SELECT, INSERT, UPDATE, DELETE com vinculação de parâmetros e processamento de resultados
Dificuldade
5/10
Tempo estimado
30 min
Etiquetas
python, web, database, sql
Pré-requisitos
Intermediate Python, SQL knowledge, sqlite3 module
python
# Web Python SQL Query Examples
# Executing SQL queries with proper parameter binding and result handling
# 1. Basic Query Execution
import sqlite3
from typing import List, Tuple, Dict, Any, Optional
def execute_query(conn: sqlite3.Connection, query: str, params: Tuple = ()) -> bool:
"""
Execute a query without returning results
Args:
conn: Connection object
query: SQL query
params: Query parameters
Returns:
True if successful
"""
try:
cursor = conn.cursor()
cursor.execute(query, params)
conn.commit()
return True
except Exception as e:
print(f"Query error: {e}")
conn.rollback()
return False
def fetch_one(conn: sqlite3.Connection, query: str, params: Tuple = ()) -> Optional[Tuple]:
"""
Fetch one row
Args:
conn: Connection object
query: SQL query
params: Query parameters
Returns:
Row tuple or None
"""
try:
cursor = conn.cursor()
cursor.execute(query, params)
return cursor.fetchone()
except Exception as e:
print(f"Fetch error: {e}")
return None
def fetch_all(conn: sqlite3.Connection, query: str, params: Tuple = ()) -> List[Tuple]:
"""
Fetch all rows
Args:
conn: Connection object
query: SQL query
params: Query parameters
Returns:
List of row tuples
"""
try:
cursor = conn.cursor()
cursor.execute(query, params)
return cursor.fetchall()
except Exception as e:
print(f"Fetch error: {e}")
return []
def fetch_many(conn: sqlite3.Connection, query: str, size: int, params: Tuple = ()) -> List[Tuple]:
"""
Fetch multiple rows
Args:
conn: Connection object
query: SQL query
size: Number of rows
params: Query parameters
Returns:
List of row tuples
"""
try:
cursor = conn.cursor()
cursor.execute(query, params)
return cursor.fetchmany(size)
except Exception as e:
print(f"Fetch error: {e}")
return []
# 2. INSERT Operations
def insert_single(conn: sqlite3.Connection, table: str, data: Dict[str, Any]) -> bool:
"""
Insert single row
Args:
conn: Connection object
table: Table name
data: Column-value pairs
Returns:
True if successful
"""
try:
columns = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
query = f'INSERT INTO {table} ({columns}) VALUES ({placeholders})'
return execute_query(conn, query, tuple(data.values()))
except Exception as e:
print(f"Insert error: {e}")
return False
def insert_many(conn: sqlite3.Connection, table: str, columns: List[str], rows: List[Tuple]) -> int:
"""
Insert multiple rows
Args:
conn: Connection object
table: Table name
columns: Column names
rows: List of value tuples
Returns:
Number of rows inserted
"""
try:
cursor = conn.cursor()
placeholders = ', '.join(['?'] * len(columns))
query = f'INSERT INTO {table} ({", ".join(columns)}) VALUES ({placeholders})'
cursor.executemany(query, rows)
conn.commit()
return cursor.rowcount
except Exception as e:
print(f"Batch insert error: {e}")
conn.rollback()
return 0
def insert_or_replace(conn: sqlite3.Connection, table: str, data: Dict[str, Any]) -> bool:
"""
Insert or replace on conflict
Args:
conn: Connection object
table: Table name
data: Column-value pairs
Returns:
True if successful
"""
try:
columns = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
query = f'INSERT OR REPLACE INTO {table} ({columns}) VALUES ({placeholders})'
return execute_query(conn, query, tuple(data.values()))
except Exception as e:
print(f"Insert or replace error: {e}")
return False
def insert_or_ignore(conn: sqlite3.Connection, table: str, data: Dict[str, Any]) -> bool:
"""
Insert or ignore on conflict
Args:
conn: Connection object
table: Table name
data: Column-value pairs
Returns:
True if successful
"""
try:
columns = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
query = f'INSERT OR IGNORE INTO {table} ({columns}) VALUES ({placeholders})'
return execute_query(conn, query, tuple(data.values()))
except Exception as e:
print(f"Insert or ignore error: {e}")
return False
def get_last_insert_id(conn: sqlite3.Connection) -> int:
"""
Get last insert row ID
Args:
conn: Connection object
Returns:
Last insert ID
"""
cursor = conn.cursor()
cursor.execute('SELECT last_insert_rowid()')
return cursor.fetchone()[0]
# 3. SELECT Operations
def select_all(conn: sqlite3.Connection, table: str, where: str = '', params: Tuple = ()) -> List[Tuple]:
"""
Select all rows
Args:
conn: Connection object
table: Table name
where: WHERE clause
params: Query parameters
Returns:
List of rows
"""
query = f'SELECT * FROM {table}'
if where:
query += f' WHERE {where}'
return fetch_all(conn, query, params)
def select_by_id(conn: sqlite3.Connection, table: str, row_id: int) -> Optional[Tuple]:
"""
Select row by ID
Args:
conn: Connection object
table: Table name
row_id: Row ID
Returns:
Row tuple or None
"""
return fetch_one(conn, f'SELECT * FROM {table} WHERE id = ?', (row_id,))
def select_columns(conn: sqlite3.Connection, table: str, columns: List[str], where: str = '', params: Tuple = ()) -> List[Tuple]:
"""
Select specific columns
Args:
conn: Connection object
table: Table name
columns: Column names
where: WHERE clause
params: Query parameters
Returns:
List of rows
"""
query = f'SELECT {", ".join(columns)} FROM {table}'
if where:
query += f' WHERE {where}'
return fetch_all(conn, query, params)
def select_with_join(conn: sqlite3.Connection, table1: str, table2: str, on: str, columns: str = '*') -> List[Tuple]:
"""
Select with JOIN
Args:
conn: Connection object
table1: First table
table2: Second table
on: JOIN condition
columns: Columns to select
Returns:
List of rows
"""
query = f'SELECT {columns} FROM {table1} JOIN {table2} ON {on}'
return fetch_all(conn, query)
def select_distinct(conn: sqlite3.Connection, table: str, column: str) -> List[Any]:
"""
Select distinct values
Args:
conn: Connection object
table: Table name
column: Column name
Returns:
List of distinct values
"""
query = f'SELECT DISTINCT {column} FROM {table}'
rows = fetch_all(conn, query)
return [row[0] for row in rows]
# 4. UPDATE Operations
def update_by_id(conn: sqlite3.Connection, table: str, row_id: int, data: Dict[str, Any]) -> bool:
"""
Update row by ID
Args:
conn: Connection object
table: Table name
row_id: Row ID
data: Column-value pairs
Returns:
True if successful
"""
try:
set_clause = ', '.join([f'{k} = ?' for k in data.keys()])
query = f'UPDATE {table} SET {set_clause} WHERE id = ?'
params = tuple(data.values()) + (row_id,)
return execute_query(conn, query, params)
except Exception as e:
print(f"Update error: {e}")
return False
def update_where(conn: sqlite3.Connection, table: str, data: Dict[str, Any], where: str, params: Tuple = ()) -> int:
"""
Update rows matching condition
Args:
conn: Connection object
table: Table name
data: Column-value pairs
where: WHERE clause
params: Additional parameters
Returns:
Number of rows updated
"""
try:
set_clause = ', '.join([f'{k} = ?' for k in data.keys()])
query = f'UPDATE {table} SET {set_clause} WHERE {where}'
cursor = conn.cursor()
cursor.execute(query, tuple(data.values()) + params)
conn.commit()
return cursor.rowcount
except Exception as e:
print(f"Update where error: {e}")
conn.rollback()
return 0
def increment_column(conn: sqlite3.Connection, table: str, column: str, where: str = '', params: Tuple = ()) -> bool:
"""
Increment column value
Args:
conn: Connection object
table: Table name
column: Column to increment
where: WHERE clause
params: Query parameters
Returns:
True if successful
"""
query = f'UPDATE {table} SET {column} = {column} + 1'
if where:
query += f' WHERE {where}'
return execute_query(conn, query, params)
# 5. DELETE Operations
def delete_by_id(conn: sqlite3.Connection, table: str, row_id: int) -> bool:
"""
Delete row by ID
Args:
conn: Connection object
table: Table name
row_id: Row ID
Returns:
True if successful
"""
return execute_query(conn, f'DELETE FROM {table} WHERE id = ?', (row_id,))
def delete_where(conn: sqlite3.Connection, table: str, where: str, params: Tuple = ()) -> int:
"""
Delete rows matching condition
Args:
conn: Connection object
table: Table name
where: WHERE clause
params: Query parameters
Returns:
Number of rows deleted
"""
try:
query = f'DELETE FROM {table} WHERE {where}'
cursor = conn.cursor()
cursor.execute(query, params)
conn.commit()
return cursor.rowcount
except Exception as e:
print(f"Delete where error: {e}")
conn.rollback()
return 0
def delete_all(conn: sqlite3.Connection, table: str) -> bool:
"""
Delete all rows from table
Args:
conn: Connection object
table: Table name
Returns:
True if successful
"""
return execute_query(conn, f'DELETE FROM {table}')
# 6. Aggregation Queries
def count_rows(conn: sqlite3.Connection, table: str, where: str = '', params: Tuple = ()) -> int:
"""
Count rows
Args:
conn: Connection object
table: Table name
where: WHERE clause
params: Query parameters
Returns:
Row count
"""
query = f'SELECT COUNT(*) FROM {table}'
if where:
query += f' WHERE {where}'
result = fetch_one(conn, query, params)
return result[0] if result else 0
def sum_column(conn: sqlite3.Connection, table: str, column: str, where: str = '') -> float:
"""
Sum column values
Args:
conn: Connection object
table: Table name
column: Column name
where: WHERE clause
Returns:
Sum value
"""
query = f'SELECT SUM({column}) FROM {table}'
if where:
query += f' WHERE {where}'
result = fetch_one(conn, query)
return result[0] if result and result[0] else 0
def avg_column(conn: sqlite3.Connection, table: str, column: str, where: str = '') -> float:
"""
Average column values
Args:
conn: Connection object
table: Table name
column: Column name
where: WHERE clause
Returns:
Average value
"""
query = f'SELECT AVG({column}) FROM {table}'
if where:
query += f' WHERE {where}'
result = fetch_one(conn, query)
return result[0] if result and result[0] else 0
def min_max_column(conn: sqlite3.Connection, table: str, column: str, where: str = '') -> Tuple:
"""
Get min and max of column
Args:
conn: Connection object
table: Table name
column: Column name
where: WHERE clause
Returns:
(min, max) tuple
"""
query = f'SELECT MIN({column}), MAX({column}) FROM {table}'
if where:
query += f' WHERE {where}'
result = fetch_one(conn, query)
return result if result else (None, None)
def group_by(conn: sqlite3.Connection, table: str, group_column: str, agg_column: str, operation: str = 'COUNT') -> List[Tuple]:
"""
Group by column
Args:
conn: Connection object
table: Table name
group_column: Column to group by
agg_column: Column to aggregate
operation: Aggregation operation
Returns:
List of (group, aggregated_value) tuples
"""
query = f'SELECT {group_column}, {operation}({agg_column}) FROM {table} GROUP BY {group_column}'
return fetch_all(conn, query)
# 7. Advanced Queries
def select_with_limit(conn: sqlite3.Connection, table: str, limit: int, offset: int = 0) -> List[Tuple]:
"""
Select with LIMIT and OFFSET
Args:
conn: Connection object
table: Table name
limit: Max rows
offset: Row offset
Returns:
List of rows
"""
query = f'SELECT * FROM {table} LIMIT {limit} OFFSET {offset}'
return fetch_all(conn, query)
def select_with_order(conn: sqlite3.Connection, table: str, order_column: str, ascending: bool = True, limit: int = 0) -> List[Tuple]:
"""
Select with ORDER BY
Args:
conn: Connection object
table: Table name
order_column: Column to order by
ascending: Sort direction
limit: Max rows
Returns:
List of rows
"""
direction = 'ASC' if ascending else 'DESC'
query = f'SELECT * FROM {table} ORDER BY {order_column} {direction}'
if limit > 0:
query += f' LIMIT {limit}'
return fetch_all(conn, query)
def search_like(conn: sqlite3.Connection, table: str, column: str, pattern: str) -> List[Tuple]:
"""
Search with LIKE
Args:
conn: Connection object
table: Table name
column: Column to search
pattern: Search pattern
Returns:
List of matching rows
"""
query = f'SELECT * FROM {table} WHERE {column} LIKE ?'
return fetch_all(conn, query, (pattern,))
def select_between(conn: sqlite3.Connection, table: str, column: str, start: Any, end: Any) -> List[Tuple]:
"""
Select with BETWEEN
Args:
conn: Connection object
table: Table name
column: Column name
start: Start value
end: End value
Returns:
List of rows
"""
query = f'SELECT * FROM {table} WHERE {column} BETWEEN ? AND ?'
return fetch_all(conn, query, (start, end))
# 8. Query Result Processing
def rows_to_dicts(rows: List[Tuple], columns: List[str]) -> List[Dict[str, Any]]:
"""
Convert rows to list of dictionaries
Args:
rows: List of row tuples
columns: Column names
Returns:
List of dictionaries
"""
return [dict(zip(columns, row)) for row in rows]
def get_row_count(conn: sqlite3.Connection, table: str) -> int:
"""Get total row count"""
return count_rows(conn, table)
def table_exists_with_data(conn: sqlite3.Connection, table: str) -> bool:
"""Check if table has data"""
return count_rows(conn, table) > 0
# Usage Examples
def demonstrate_sqlite_query():
print("=== Web Python SQL Query Examples ===\n")
# Setup database
conn = sqlite3.connect(':memory:')
conn.execute('''
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT,
age INTEGER,
salary REAL
)
''')
conn.commit()
# 1. INSERT operations
print("--- 1. INSERT Operations ---")
insert_single(conn, 'users', {'name': 'Alice', 'email': '[email protected]', 'age': 30, 'salary': 50000})
insert_single(conn, 'users', {'name': 'Bob', 'email': '[email protected]', 'age': 25, 'salary': 45000})
insert_many(conn, 'users', ['name', 'email', 'age', 'salary'], [
('Charlie', '[email protected]', 35, 60000),
('David', '[email protected]', 28, 55000),
('Eve', '[email protected]', 32, 58000)
])
print(f"Rows inserted: {count_rows(conn, 'users')}")
print(f"Last insert ID: {get_last_insert_id(conn)}")
# 2. SELECT operations
print("\n--- 2. SELECT Operations ---")
print(f"All users: {fetch_all(conn, 'SELECT * FROM users')}")
print(f"User by ID 1: {fetch_one(conn, 'SELECT * FROM users WHERE id = ?', (1,))}")
print(f"Select columns: {select_columns(conn, 'users', ['name', 'email'])}")
# 3. UPDATE operations
print("\n--- 3. UPDATE Operations ---")
update_by_id(conn, 'users', 1, {'salary': 55000})
increment_column(conn, 'users', 'age', 'name = ?', ('Alice',))
print(f"After update: {fetch_one(conn, 'SELECT * FROM users WHERE id = 1')}")
# 4. DELETE operations
print("\n--- 4. DELETE Operations ---")
delete_by_id(conn, 'users', 5)
print(f"After delete ID 5: {count_rows(conn, 'users')}")
# 5. Aggregation
print("\n--- 5. Aggregation Queries ---")
print(f"Count: {count_rows(conn, 'users')}")
print(f"Average salary: {avg_column(conn, 'users', 'salary'):.2f}")
print(f"Total salary: {sum_column(conn, 'users', 'salary'):.2f}")
print(f"Min/max age: {min_max_column(conn, 'users', 'age')}")
print(f"Group by age range: {group_by(conn, 'users', 'age', 'salary', 'AVG')}")
# 6. Advanced queries
print("\n--- 6. Advanced Queries ---")
print(f"Limit 2: {select_with_limit(conn, 'users', 2)}")
print(f"Order by salary: {select_with_order(conn, 'users', 'salary', ascending=False)}")
print(f"Like search: {search_like(conn, 'users', 'name', '%a%')}")
# 7. BETWEEN
print("\n--- 7. BETWEEN Query ---")
print(f"Age between 25 and 32: {select_between(conn, 'users', 'age', 25, 32)}")
conn.close()
print("\n=== All SQL Query Examples Completed ===")
# Export functions
# export { execute_query, fetch_one, fetch_all, fetch_many }
# export { insert_single, insert_many, insert_or_replace, insert_or_ignore, get_last_insert_id }
# export { select_all, select_by_id, select_columns, select_with_join, select_distinct }
# export { update_by_id, update_where, increment_column }
# export { delete_by_id, delete_where, delete_all }
# export { count_rows, sum_column, avg_column, min_max_column, group_by }
# export { select_with_limit, select_with_order, search_like, select_between }
# export { rows_to_dicts, get_row_count, table_exists_with_data }
# export { demonstrate_sqlite_query }
CódigopythonIntermediário
Processamento de Transações
Manipular transações de banco de dados com operações de commit, rollback e savepoint
Dificuldade
6/10
Tempo estimado
35 min
Etiquetas
python, web, database, transaction
Pré-requisitos
Intermediate Python, Database concepts, ACID properties
python
# Web Python Transaction Examples
# Managing database transactions with ACID properties
# 1. Basic Transaction Control
import sqlite3
from typing import Optional, List, Callable, Any
from contextlib import contextmanager
def begin_transaction(conn: sqlite3.Connection) -> None:
"""
Begin explicit transaction
Args:
conn: Connection object
"""
conn.execute('BEGIN TRANSACTION')
def commit_transaction(conn: sqlite3.Connection) -> bool:
"""
Commit transaction
Args:
conn: Connection object
Returns:
True if successful
"""
try:
conn.commit()
return True
except Exception as e:
print(f"Commit error: {e}")
return False
def rollback_transaction(conn: sqlite3.Connection) -> bool:
"""
Rollback transaction
Args:
conn: Connection object
Returns:
True if successful
"""
try:
conn.rollback()
return True
except Exception as e:
print(f"Rollback error: {e}")
return False
def get_transaction_status(conn: sqlite3.Connection) -> str:
"""
Get transaction status
Args:
conn: Connection object
Returns:
Transaction status string
"""
# Check if in transaction
cursor = conn.cursor()
cursor.execute('BEGIN')
try:
cursor.execute('COMMIT')
return 'no transaction'
except:
return 'in transaction'
# 2. Context Manager Transactions
@contextmanager
def transaction(conn: sqlite3.Connection):
"""
Context manager for transactions
Args:
conn: Connection object
Yields:
Connection object
Example:
with transaction(conn):
# operations
pass # auto commit on success, rollback on error
"""
try:
begin_transaction(conn)
yield conn
conn.commit()
except Exception as e:
conn.rollback()
raise e
def execute_in_transaction(conn: sqlite3.Connection, operations: List[Callable]) -> bool:
"""
Execute multiple operations in transaction
Args:
conn: Connection object
operations: List of callable operations
Returns:
True if all successful
"""
try:
with transaction(conn):
for operation in operations:
operation(conn)
return True
except Exception as e:
print(f"Transaction failed: {e}")
return False
# 3. Savepoint Operations
def create_savepoint(conn: sqlite3.Connection, name: str) -> None:
"""
Create savepoint
Args:
conn: Connection object
name: Savepoint name
"""
conn.execute(f'SAVEPOINT {name}')
def release_savepoint(conn: sqlite3.Connection, name: str) -> bool:
"""
Release savepoint
Args:
conn: Connection object
name: Savepoint name
Returns:
True if successful
"""
try:
conn.execute(f'RELEASE SAVEPOINT {name}')
return True
except Exception as e:
print(f"Release savepoint error: {e}")
return False
def rollback_to_savepoint(conn: sqlite3.Connection, name: str) -> bool:
"""
Rollback to savepoint
Args:
conn: Connection object
name: Savepoint name
Returns:
True if successful
"""
try:
conn.execute(f'ROLLBACK TO SAVEPOINT {name}')
return True
except Exception as e:
print(f"Rollback to savepoint error: {e}")
return False
@contextmanager
def savepoint(conn: sqlite3.Connection, name: str):
"""
Context manager for savepoint
Args:
conn: Connection object
name: Savepoint name
Yields:
Connection object
"""
try:
create_savepoint(conn, name)
yield conn
release_savepoint(conn, name)
except Exception as e:
rollback_to_savepoint(conn, name)
raise e
# 4. Isolation Levels
def set_isolation_level(conn: sqlite3.Connection, level: str) -> None:
"""
Set isolation level
Args:
conn: Connection object
level: 'DEFERRED', 'IMMEDIATE', 'EXCLUSIVE', or None (autocommit)
"""
conn.isolation_level = level
def get_isolation_level(conn: sqlite3.Connection) -> str:
"""
Get current isolation level
Args:
conn: Connection object
Returns:
Isolation level string
"""
return str(conn.isolation_level)
def deferred_transaction(conn: sqlite3.Connection) -> None:
"""Begin deferred transaction (default)"""
conn.execute('BEGIN DEFERRED')
def immediate_transaction(conn: sqlite3.Connection) -> None:
"""Begin immediate transaction (reserves reserved locks)"""
conn.execute('BEGIN IMMEDIATE')
def exclusive_transaction(conn: sqlite3.Connection) -> None:
"""Begin exclusive transaction (reserves all locks)"""
conn.execute('BEGIN EXCLUSIVE')
# 5. Transaction Patterns
def transfer_funds(conn: sqlite3.Connection, from_account: int, to_account: int, amount: float) -> bool:
"""
Transfer funds between accounts (atomic operation)
Args:
conn: Connection object
from_account: Source account ID
to_account: Destination account ID
amount: Transfer amount
Returns:
True if successful
"""
try:
with transaction(conn):
# Deduct from source
conn.execute(
'UPDATE accounts SET balance = balance - ? WHERE id = ?',
(amount, from_account)
)
# Add to destination
conn.execute(
'UPDATE accounts SET balance = balance + ? WHERE id = ?',
(amount, to_account)
)
# Verify sufficient funds
cursor = conn.execute(
'SELECT balance FROM accounts WHERE id = ?',
(from_account,)
)
balance = cursor.fetchone()[0]
if balance < 0:
raise Exception("Insufficient funds")
return True
except Exception as e:
print(f"Transfer failed: {e}")
return False
def batch_insert(conn: sqlite3.Connection, table: str, rows: List[tuple], batch_size: int = 1000) -> int:
"""
Batch insert with transaction
Args:
conn: Connection object
table: Table name
rows: List of row tuples
batch_size: Rows per transaction
Returns:
Total rows inserted
"""
total_inserted = 0
for i in range(0, len(rows), batch_size):
batch = rows[i:i + batch_size]
try:
with transaction(conn):
cursor = conn.cursor()
cursor.executemany(f'INSERT INTO {table} VALUES ({", ".join(["?"]*len(batch[0]))})', batch)
total_inserted += cursor.rowcount
except Exception as e:
print(f"Batch insert failed at {i}: {e}")
break
return total_inserted
def upsert_record(conn: sqlite3.Connection, table: str, data: dict, key_column: str) -> bool:
"""
Insert or update record
Args:
conn: Connection object
table: Table name
data: Record data
key_column: Column to check for existence
Returns:
True if successful
"""
try:
with transaction(conn):
# Check if exists
cursor = conn.execute(f'SELECT id FROM {table} WHERE {key_column} = ?', (data[key_column],))
if cursor.fetchone():
# Update
set_clause = ', '.join([f'{k} = ?' for k in data.keys()])
conn.execute(f'UPDATE {table} SET {set_clause} WHERE {key_column} = ?', tuple(data.values()) + (data[key_column],))
else:
# Insert
columns = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
conn.execute(f'INSERT INTO {table} ({columns}) VALUES ({placeholders})', tuple(data.values()))
return True
except Exception as e:
print(f"Upsert failed: {e}")
return False
# 6. Nested Transactions
def nested_transaction_example(conn: sqlite3.Connection) -> bool:
"""
Example of nested transactions with savepoints
Args:
conn: Connection object
Returns:
True if successful
"""
try:
with transaction(conn):
# Outer transaction operations
conn.execute('INSERT INTO users (name) VALUES (?)', ('Alice',))
# Nested savepoint
with savepoint(conn, 'nested'):
conn.execute('INSERT INTO orders (user_id) VALUES (?)', (1,))
# Can rollback to this savepoint if needed
return True
except Exception as e:
print(f"Nested transaction failed: {e}")
return False
# 7. Transaction Logging
class TransactionLogger:
"""Log all transaction operations"""
def __init__(self, conn: sqlite3.Connection):
self.conn = conn
self.log: List[str] = []
def execute(self, query: str, params: tuple = ()):
"""Execute with logging"""
self.log.append(f"EXECUTE: {query} with {params}")
return self.conn.execute(query, params)
def commit(self):
"""Commit with logging"""
self.log.append("COMMIT")
return self.conn.commit()
def rollback(self):
"""Rollback with logging"""
self.log.append("ROLLBACK")
return self.conn.rollback()
def get_log(self) -> List[str]:
"""Get operation log"""
return self.log
# 8. Atomic Operations
def atomic_operation(conn: sqlite3.Connection, operation: Callable, *args, **kwargs) -> Any:
"""
Execute operation atomically
Args:
conn: Connection object
operation: Callable to execute
*args: Operation arguments
**kwargs: Operation keyword arguments
Returns:
Operation result or None on failure
"""
try:
with transaction(conn):
return operation(conn, *args, **kwargs)
except Exception as e:
print(f"Atomic operation failed: {e}")
return None
def compare_and_update(conn: sqlite3.Connection, table: str, id: int, column: str, old_value: Any, new_value: Any) -> bool:
"""
Atomic compare-and-update
Args:
conn: Connection object
table: Table name
id: Row ID
column: Column name
old_value: Expected current value
new_value: New value
Returns:
True if updated (value matched)
"""
try:
with transaction(conn):
cursor = conn.execute(f'SELECT {column} FROM {table} WHERE id = ?', (id,))
current = cursor.fetchone()[0]
if current == old_value:
conn.execute(f'UPDATE {table} SET {column} = ? WHERE id = ?', (new_value, id))
return True
return False
except Exception as e:
print(f"Compare and update failed: {e}")
return False
# 9. Transaction State Management
def in_transaction(conn: sqlite3.Connection) -> bool:
"""
Check if connection is in transaction
Args:
conn: Connection object
Returns:
True if in transaction
"""
return conn.in_transaction
def get_changes_count(conn: sqlite3.Connection) -> int:
"""
Get number of changes in transaction
Args:
conn: Connection object
Returns:
Number of changes
"""
return conn.total_changes
def reset_transaction(conn: sqlite3.Connection) -> None:
"""
Reset transaction state
Args:
conn: Connection object
"""
conn.rollback()
# Usage Examples
def demonstrate_transaction():
print("=== Web Python Transaction Examples ===\n")
# Setup database
conn = sqlite3.connect(':memory:')
conn.execute('''
CREATE TABLE accounts (
id INTEGER PRIMARY KEY,
name TEXT,
balance REAL
)
''')
conn.execute('''
CREATE TABLE users (
id INTEGER PRIMARY KEY,
name TEXT
)
''')
conn.execute('''
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
user_id INTEGER
)
''')
conn.commit()
# Insert initial data
conn.execute('INSERT INTO accounts VALUES (1, "Alice", 1000)')
conn.execute('INSERT INTO accounts VALUES (2, "Bob", 500)')
conn.commit()
# 1. Basic transaction
print("--- 1. Basic Transaction ---")
try:
with transaction(conn):
conn.execute('UPDATE accounts SET balance = balance + 100 WHERE id = 1')
print("In transaction: updated, not committed yet")
# Auto commit on success
except Exception as e:
print(f"Transaction failed: {e}")
print(f"After transaction: {fetch_all(conn, 'SELECT * FROM accounts')}")
# 2. Rollback example
print("\n--- 2. Rollback Example ---")
try:
with transaction(conn):
conn.execute('UPDATE accounts SET balance = balance - 1000 WHERE id = 1')
raise Exception("Intentional error")
except:
pass
print(f"After rollback: {fetch_all(conn, 'SELECT * FROM accounts')}")
# 3. Savepoints
print("\n--- 3. Savepoints ---")
try:
with transaction(conn):
conn.execute('UPDATE accounts SET balance = balance + 50 WHERE id = 1')
with savepoint(conn, 'sp1'):
conn.execute('UPDATE accounts SET balance = balance + 50 WHERE id = 2')
print("After nested update")
# Outer transaction continues
except Exception as e:
print(f"Error: {e}")
print(f"After savepoint: {fetch_all(conn, 'SELECT * FROM accounts')}")
# 4. Transfer funds
print("\n--- 4. Transfer Funds (Atomic) ---")
print(f"Before transfer: {fetch_all(conn, 'SELECT * FROM accounts')}")
success = transfer_funds(conn, 1, 2, 200)
print(f"Transfer success: {success}")
print(f"After transfer: {fetch_all(conn, 'SELECT * FROM accounts')}")
# 5. Batch insert
print("\n--- 5. Batch Insert ---")
batch_data = [
('User1',),
('User2',),
('User3',),
]
inserted = batch_insert(conn, 'users', batch_data, batch_size=2)
print(f"Batch inserted {inserted} rows")
print(f"Users: {fetch_all(conn, 'SELECT * FROM users')}")
# 6. Upsert
print("\n--- 6. Upsert ---")
upsert_record(conn, 'users', {'id': 4, 'name': 'User4'}, 'id')
upsert_record(conn, 'users', {'id': 1, 'name': 'UpdatedUser1'}, 'id')
print(f"After upsert: {fetch_all(conn, 'SELECT * FROM users')}")
# 7. Compare and update
print("\n--- 7. Compare and Update ---")
result = compare_and_update(conn, 'accounts', 1, 'balance', 1150, 1200)
print(f"Compare and update success: {result}")
print(f"Accounts: {fetch_all(conn, 'SELECT * FROM accounts')}")
# 8. Transaction logging
print("\n--- 8. Transaction Logging ---")
logger = TransactionLogger(conn)
try:
with transaction(conn):
logger.execute('UPDATE accounts SET balance = balance + 10 WHERE id = 1')
logger.execute('UPDATE accounts SET balance = balance - 10 WHERE id = 2')
except:
pass
print(f"Transaction log: {logger.get_log()}")
conn.close()
print("\n=== All Transaction Examples Completed ===")
def fetch_all(conn, query, params=()):
"""Helper function"""
cursor = conn.cursor()
cursor.execute(query, params)
return cursor.fetchall()
# Export functions
# export { begin_transaction, commit_transaction, rollback_transaction, get_transaction_status }
# export { transaction, execute_in_transaction }
# export { create_savepoint, release_savepoint, rollback_to_savepoint, savepoint }
# export { set_isolation_level, get_isolation_level, deferred_transaction, immediate_transaction, exclusive_transaction }
# export { transfer_funds, batch_insert, upsert_record, nested_transaction_example }
# export { TransactionLogger }
# export { atomic_operation, compare_and_update }
# export { in_transaction, get_changes_count, reset_transaction }
# export { demonstrate_transaction }
Ferramentas
Ferramentas frequentemente usadas com este exemplo
Relacionado