udbcore 1.1.1

This package provides foundational data structures for representing and manipulating tile maps in 2D and 3D environments. Its primary purpose is to enable efficient spatial organization and management of map data for games, simulations, and applications that require robust handling of coordinates, regions, and chunked regions. By offering specialized classes for coordinates, regions, and chunked regions, the package simplifies the development of systems that need precise and scalable map logic.

udbcore

The udbcore submodule provides comprehensive database abstractions and utilities for working with DuckDB databases in this project. It offers a clean, robust interface for database management with advanced features like connection pooling, transaction management, query building, schema management, and health monitoring.

Contents

Core Components

• db.py: Defines the DB base class for generic database file management
• ddb.py: Defines the enhanced DuckDB class with retry logic, health monitoring, and advanced operations

Advanced Features

• connection_pool.py: Connection pooling utilities for efficient database connection management
• transactions.py: Transaction management and batch operation utilities
• query_builder.py: Fluent SQL query builder with support for complex queries
• schema.py: Database schema management utilities for table/index creation and migration
• monitoring.py: Database health monitoring and performance statistics

Quick Start

from foundation_packages.udbcore import DuckDB

# Create a DuckDB database object
my_db = DuckDB(path='path/to/database.duckdb', name='database.duckdb')

# Connect and run a query
my_db.connect()
results = my_db.run_query('SELECT * FROM my_table;')
my_db.disconnect()

Core Features

Enhanced DuckDB Class

The DuckDB class provides robust database operations with automatic retry logic and health monitoring:

# Initialize with retry configuration
db = DuckDB('data.duckdb', 'data.duckdb', retry_attempts=3, retry_delay=1.0)

# Execute queries with automatic retry
results = db.run_query("SELECT * FROM users WHERE age > ?", [18])
single_result = db.run_query_single("SELECT COUNT(*) FROM users")

# Execute non-query operations
rows_affected = db.execute_non_query("INSERT INTO users (name, age) VALUES (?, ?)", ['John', 25])

# Batch operations
db.execute_many("INSERT INTO users (name, age) VALUES (?, ?)", 
                [['Alice', 30], ['Bob', 35], ['Carol', 28]])

# Transaction support
with db.transaction():
    db.execute_non_query("INSERT INTO accounts (user_id, balance) VALUES (?, ?)", [1, 100])
    db.execute_non_query("UPDATE users SET account_created = true WHERE id = ?", [1])

Connection Pooling

Efficient connection management for multiple databases:

from foundation_packages.udbcore import ConnectionPool, get_global_pool

# Use global connection pool
pool = get_global_pool()
db = pool.get_connection('/path/to/database.duckdb')
results = db.run_query("SELECT * FROM my_table")
pool.return_connection('/path/to/database.duckdb')

# Create custom pool
custom_pool = ConnectionPool(max_connections=20, connection_timeout=600)
db = custom_pool.get_connection('/path/to/another_db.duckdb')

Transaction Management

Advanced transaction control with automatic rollback:

from foundation_packages.udbcore import TransactionManager, BatchOperationManager

# Transaction management
db = DuckDB('data.duckdb', 'data.duckdb')
tm = TransactionManager(db)

# Manual transaction control
tm.begin_transaction()
try:
    db.execute_non_query("INSERT INTO table1 VALUES (?)", [value1])
    db.execute_non_query("INSERT INTO table2 VALUES (?)", [value2])
    tm.commit()
except Exception:
    tm.rollback()

# Context manager (automatic)
with tm.transaction():
    db.execute_non_query("INSERT INTO table1 VALUES (?)", [value1])
    db.execute_non_query("INSERT INTO table2 VALUES (?)", [value2])

# Batch operations
bm = BatchOperationManager(db, batch_size=1000)
data = [['Alice', 25], ['Bob', 30], ['Carol', 28]]
rows_inserted = bm.batch_insert('users', ['name', 'age'], data)

Query Builder

Fluent interface for building complex SQL queries:

from foundation_packages.udbcore import QueryBuilder, JoinType, OrderDirection

qb = QueryBuilder()

# Simple query
query = (qb.reset()
         .select('name', 'age', 'email')
         .from_table('users')
         .where('age > 18')
         .order_by('name', OrderDirection.ASC)
         .limit(10)
         .build())

# Complex query with joins
query = (qb.reset()
         .select('u.name', 'p.title', 'c.name')
         .from_table('users u')
         .join('posts p', 'p.user_id = u.id', JoinType.LEFT)
         .join('categories c', 'c.id = p.category_id', JoinType.INNER)
         .where('u.active = true')
         .where_in('c.name', ['Tech', 'Science'])
         .group_by('u.id')
         .having('COUNT(p.id) > 5')
         .order_by('u.name')
         .build())

# Execute the query
results = db.run_query(query)

Schema Management

Programmatic database schema creation and management:

from foundation_packages.udbcore import (SchemaManager, TableDefinition, 
                                       ColumnDefinition, IndexDefinition, ColumnType)

# Initialize schema manager
db = DuckDB('data.duckdb', 'data.duckdb')
sm = SchemaManager(db)

# Define table structure
columns = [
    ColumnDefinition('id', ColumnType.INTEGER, primary_key=True),
    ColumnDefinition('name', ColumnType.VARCHAR, nullable=False),
    ColumnDefinition('email', ColumnType.VARCHAR, unique=True),
    ColumnDefinition('age', ColumnType.INTEGER, default=0),
    ColumnDefinition('created_at', ColumnType.TIMESTAMP, default='CURRENT_TIMESTAMP')
]

indexes = [
    IndexDefinition('idx_users_email', 'users', ['email'], unique=True),
    IndexDefinition('idx_users_age', 'users', ['age'])
]

table_def = TableDefinition('users', columns, indexes)

# Create table
sm.create_table(table_def)

# Schema operations
sm.add_column('users', ColumnDefinition('last_login', ColumnType.TIMESTAMP))
sm.create_index(IndexDefinition('idx_users_name', 'users', ['name']))

# Schema inspection
tables = sm.list_tables()
schema_info = sm.get_table_schema('users')
exists = sm.table_exists('users')

Health Monitoring

Comprehensive database health monitoring and performance statistics:

from foundation_packages.udbcore import HealthMonitor

# Get health monitor from database
db = DuckDB('data.duckdb', 'data.duckdb')
monitor = db.get_health_monitor()

# Get statistics
query_stats = monitor.get_query_statistics()
db_stats = monitor.get_database_statistics()
table_stats = monitor.get_table_statistics()

# Health check
health = monitor.health_check()
print(f"Database status: {health['overall_status']}")

# Export comprehensive stats
all_stats = monitor.export_statistics()

# Find slow queries
slow_queries = monitor.get_slow_queries(threshold_seconds=2.0)

Advanced Usage Examples

Tilemap Region Database Management

from foundation_packages.udbcore import DuckDB, SchemaManager, ColumnDefinition, ColumnType

# Create region database with proper schema
region_db = DuckDB(f'regions/region_{rx}_{ry}.duckdb', f'region_{rx}_{ry}.duckdb')
schema_manager = SchemaManager(region_db)

# Define cells table schema
cells_columns = [
    ColumnDefinition('x', ColumnType.INTEGER, nullable=False),
    ColumnDefinition('y', ColumnType.INTEGER, nullable=False),
    ColumnDefinition('z', ColumnType.INTEGER, nullable=False),
    ColumnDefinition('tile', ColumnType.INTEGER, default=0),
    ColumnDefinition('properties', ColumnType.JSON)
]

# Create spatial indexes for efficient lookups
spatial_indexes = [
    IndexDefinition('idx_cells_xyz', 'cells', ['x', 'y', 'z'], unique=True),
    IndexDefinition('idx_cells_xy', 'cells', ['x', 'y']),
    IndexDefinition('idx_cells_tile', 'cells', ['tile'])
]

table_def = TableDefinition('cells', cells_columns, spatial_indexes)
schema_manager.create_table(table_def)

# Efficient batch cell operations
batch_manager = BatchOperationManager(region_db, batch_size=10000)

# Insert many cells efficiently
cell_data = [(x, y, z, tile_id) for x in range(128) for y in range(128) for z in range(128)]
batch_manager.batch_insert('cells', ['x', 'y', 'z', 'tile'], cell_data)

Multi-Database Operations

# Use connection pool for multiple region databases
pool = get_global_pool()

regions_to_process = [(0, 0), (0, 1), (1, 0), (1, 1)]

for rx, ry in regions_to_process:
    db_path = f'regions/region_{rx}_{ry}.duckdb'
    db = pool.get_connection(db_path)
    
    # Process each region
    with db.transaction():
        # Update tiles based on some logic
        updated_tiles = db.run_query(
            "SELECT x, y, z FROM cells WHERE tile = ? AND z > ?", 
            [old_tile_id, min_height]
        )
        
        if updated_tiles:
            db.execute_non_query(
                "UPDATE cells SET tile = ? WHERE tile = ? AND z > ?",
                [new_tile_id, old_tile_id, min_height]
            )
    
    pool.return_connection(db_path)

Requirements

• duckdb >= 0.8.0

Version History

• v1.0.0 - Initial release

License

See the main project for license information.