| Name | udbcore JSON |
| Version |
1.0
JSON |
| download |
| home_page | https://github.com/uaineteine/udbcore |
| Summary | 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. |
| upload_time | 2025-09-16 14:37:36 |
| maintainer | None |
| docs_url | None |
| author | Daniel Stamer-Squair |
| requires_python | >=3.12 |
| license | None |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# 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
```python
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:
```python
# 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:
```python
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:
```python
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:
```python
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:
```python
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:
```python
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
```python
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
```python
# 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](https://pypi.org/project/duckdb/) >= 0.8.0
## Version History
- **v1.0.0** - Initial release
## License
See the main project for license information.
Raw data
{
"_id": null,
"home_page": "https://github.com/uaineteine/udbcore",
"name": "udbcore",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": null,
"author": "Daniel Stamer-Squair",
"author_email": "uaine.teine@hotmail.com",
"download_url": "https://files.pythonhosted.org/packages/8a/c0/051083c33eb5c2cbbe4c254b153b4f05957af1071d1dcebd9f826b8e029b/udbcore-1.0.tar.gz",
"platform": null,
"description": "# udbcore\r\n\r\nThe `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.\r\n\r\n## Contents\r\n\r\n### Core Components\r\n- `db.py`: Defines the `DB` base class for generic database file management\r\n- `ddb.py`: Defines the enhanced `DuckDB` class with retry logic, health monitoring, and advanced operations\r\n\r\n### Advanced Features\r\n- `connection_pool.py`: Connection pooling utilities for efficient database connection management\r\n- `transactions.py`: Transaction management and batch operation utilities\r\n- `query_builder.py`: Fluent SQL query builder with support for complex queries\r\n- `schema.py`: Database schema management utilities for table/index creation and migration\r\n- `monitoring.py`: Database health monitoring and performance statistics\r\n\r\n## Quick Start\r\n\r\n```python\r\nfrom foundation_packages.udbcore import DuckDB\r\n\r\n# Create a DuckDB database object\r\nmy_db = DuckDB(path='path/to/database.duckdb', name='database.duckdb')\r\n\r\n# Connect and run a query\r\nmy_db.connect()\r\nresults = my_db.run_query('SELECT * FROM my_table;')\r\nmy_db.disconnect()\r\n```\r\n\r\n## Core Features\r\n\r\n### Enhanced DuckDB Class\r\n\r\nThe `DuckDB` class provides robust database operations with automatic retry logic and health monitoring:\r\n\r\n```python\r\n# Initialize with retry configuration\r\ndb = DuckDB('data.duckdb', 'data.duckdb', retry_attempts=3, retry_delay=1.0)\r\n\r\n# Execute queries with automatic retry\r\nresults = db.run_query(\"SELECT * FROM users WHERE age > ?\", [18])\r\nsingle_result = db.run_query_single(\"SELECT COUNT(*) FROM users\")\r\n\r\n# Execute non-query operations\r\nrows_affected = db.execute_non_query(\"INSERT INTO users (name, age) VALUES (?, ?)\", ['John', 25])\r\n\r\n# Batch operations\r\ndb.execute_many(\"INSERT INTO users (name, age) VALUES (?, ?)\", \r\n [['Alice', 30], ['Bob', 35], ['Carol', 28]])\r\n\r\n# Transaction support\r\nwith db.transaction():\r\n db.execute_non_query(\"INSERT INTO accounts (user_id, balance) VALUES (?, ?)\", [1, 100])\r\n db.execute_non_query(\"UPDATE users SET account_created = true WHERE id = ?\", [1])\r\n```\r\n\r\n### Connection Pooling\r\n\r\nEfficient connection management for multiple databases:\r\n\r\n```python\r\nfrom foundation_packages.udbcore import ConnectionPool, get_global_pool\r\n\r\n# Use global connection pool\r\npool = get_global_pool()\r\ndb = pool.get_connection('/path/to/database.duckdb')\r\nresults = db.run_query(\"SELECT * FROM my_table\")\r\npool.return_connection('/path/to/database.duckdb')\r\n\r\n# Create custom pool\r\ncustom_pool = ConnectionPool(max_connections=20, connection_timeout=600)\r\ndb = custom_pool.get_connection('/path/to/another_db.duckdb')\r\n```\r\n\r\n### Transaction Management\r\n\r\nAdvanced transaction control with automatic rollback:\r\n\r\n```python\r\nfrom foundation_packages.udbcore import TransactionManager, BatchOperationManager\r\n\r\n# Transaction management\r\ndb = DuckDB('data.duckdb', 'data.duckdb')\r\ntm = TransactionManager(db)\r\n\r\n# Manual transaction control\r\ntm.begin_transaction()\r\ntry:\r\n db.execute_non_query(\"INSERT INTO table1 VALUES (?)\", [value1])\r\n db.execute_non_query(\"INSERT INTO table2 VALUES (?)\", [value2])\r\n tm.commit()\r\nexcept Exception:\r\n tm.rollback()\r\n\r\n# Context manager (automatic)\r\nwith tm.transaction():\r\n db.execute_non_query(\"INSERT INTO table1 VALUES (?)\", [value1])\r\n db.execute_non_query(\"INSERT INTO table2 VALUES (?)\", [value2])\r\n\r\n# Batch operations\r\nbm = BatchOperationManager(db, batch_size=1000)\r\ndata = [['Alice', 25], ['Bob', 30], ['Carol', 28]]\r\nrows_inserted = bm.batch_insert('users', ['name', 'age'], data)\r\n```\r\n\r\n### Query Builder\r\n\r\nFluent interface for building complex SQL queries:\r\n\r\n```python\r\nfrom foundation_packages.udbcore import QueryBuilder, JoinType, OrderDirection\r\n\r\nqb = QueryBuilder()\r\n\r\n# Simple query\r\nquery = (qb.reset()\r\n .select('name', 'age', 'email')\r\n .from_table('users')\r\n .where('age > 18')\r\n .order_by('name', OrderDirection.ASC)\r\n .limit(10)\r\n .build())\r\n\r\n# Complex query with joins\r\nquery = (qb.reset()\r\n .select('u.name', 'p.title', 'c.name')\r\n .from_table('users u')\r\n .join('posts p', 'p.user_id = u.id', JoinType.LEFT)\r\n .join('categories c', 'c.id = p.category_id', JoinType.INNER)\r\n .where('u.active = true')\r\n .where_in('c.name', ['Tech', 'Science'])\r\n .group_by('u.id')\r\n .having('COUNT(p.id) > 5')\r\n .order_by('u.name')\r\n .build())\r\n\r\n# Execute the query\r\nresults = db.run_query(query)\r\n```\r\n\r\n### Schema Management\r\n\r\nProgrammatic database schema creation and management:\r\n\r\n```python\r\nfrom foundation_packages.udbcore import (SchemaManager, TableDefinition, \r\n ColumnDefinition, IndexDefinition, ColumnType)\r\n\r\n# Initialize schema manager\r\ndb = DuckDB('data.duckdb', 'data.duckdb')\r\nsm = SchemaManager(db)\r\n\r\n# Define table structure\r\ncolumns = [\r\n ColumnDefinition('id', ColumnType.INTEGER, primary_key=True),\r\n ColumnDefinition('name', ColumnType.VARCHAR, nullable=False),\r\n ColumnDefinition('email', ColumnType.VARCHAR, unique=True),\r\n ColumnDefinition('age', ColumnType.INTEGER, default=0),\r\n ColumnDefinition('created_at', ColumnType.TIMESTAMP, default='CURRENT_TIMESTAMP')\r\n]\r\n\r\nindexes = [\r\n IndexDefinition('idx_users_email', 'users', ['email'], unique=True),\r\n IndexDefinition('idx_users_age', 'users', ['age'])\r\n]\r\n\r\ntable_def = TableDefinition('users', columns, indexes)\r\n\r\n# Create table\r\nsm.create_table(table_def)\r\n\r\n# Schema operations\r\nsm.add_column('users', ColumnDefinition('last_login', ColumnType.TIMESTAMP))\r\nsm.create_index(IndexDefinition('idx_users_name', 'users', ['name']))\r\n\r\n# Schema inspection\r\ntables = sm.list_tables()\r\nschema_info = sm.get_table_schema('users')\r\nexists = sm.table_exists('users')\r\n```\r\n\r\n### Health Monitoring\r\n\r\nComprehensive database health monitoring and performance statistics:\r\n\r\n```python\r\nfrom foundation_packages.udbcore import HealthMonitor\r\n\r\n# Get health monitor from database\r\ndb = DuckDB('data.duckdb', 'data.duckdb')\r\nmonitor = db.get_health_monitor()\r\n\r\n# Get statistics\r\nquery_stats = monitor.get_query_statistics()\r\ndb_stats = monitor.get_database_statistics()\r\ntable_stats = monitor.get_table_statistics()\r\n\r\n# Health check\r\nhealth = monitor.health_check()\r\nprint(f\"Database status: {health['overall_status']}\")\r\n\r\n# Export comprehensive stats\r\nall_stats = monitor.export_statistics()\r\n\r\n# Find slow queries\r\nslow_queries = monitor.get_slow_queries(threshold_seconds=2.0)\r\n```\r\n\r\n## Advanced Usage Examples\r\n\r\n### Tilemap Region Database Management\r\n\r\n```python\r\nfrom foundation_packages.udbcore import DuckDB, SchemaManager, ColumnDefinition, ColumnType\r\n\r\n# Create region database with proper schema\r\nregion_db = DuckDB(f'regions/region_{rx}_{ry}.duckdb', f'region_{rx}_{ry}.duckdb')\r\nschema_manager = SchemaManager(region_db)\r\n\r\n# Define cells table schema\r\ncells_columns = [\r\n ColumnDefinition('x', ColumnType.INTEGER, nullable=False),\r\n ColumnDefinition('y', ColumnType.INTEGER, nullable=False),\r\n ColumnDefinition('z', ColumnType.INTEGER, nullable=False),\r\n ColumnDefinition('tile', ColumnType.INTEGER, default=0),\r\n ColumnDefinition('properties', ColumnType.JSON)\r\n]\r\n\r\n# Create spatial indexes for efficient lookups\r\nspatial_indexes = [\r\n IndexDefinition('idx_cells_xyz', 'cells', ['x', 'y', 'z'], unique=True),\r\n IndexDefinition('idx_cells_xy', 'cells', ['x', 'y']),\r\n IndexDefinition('idx_cells_tile', 'cells', ['tile'])\r\n]\r\n\r\ntable_def = TableDefinition('cells', cells_columns, spatial_indexes)\r\nschema_manager.create_table(table_def)\r\n\r\n# Efficient batch cell operations\r\nbatch_manager = BatchOperationManager(region_db, batch_size=10000)\r\n\r\n# Insert many cells efficiently\r\ncell_data = [(x, y, z, tile_id) for x in range(128) for y in range(128) for z in range(128)]\r\nbatch_manager.batch_insert('cells', ['x', 'y', 'z', 'tile'], cell_data)\r\n```\r\n\r\n### Multi-Database Operations\r\n\r\n```python\r\n# Use connection pool for multiple region databases\r\npool = get_global_pool()\r\n\r\nregions_to_process = [(0, 0), (0, 1), (1, 0), (1, 1)]\r\n\r\nfor rx, ry in regions_to_process:\r\n db_path = f'regions/region_{rx}_{ry}.duckdb'\r\n db = pool.get_connection(db_path)\r\n \r\n # Process each region\r\n with db.transaction():\r\n # Update tiles based on some logic\r\n updated_tiles = db.run_query(\r\n \"SELECT x, y, z FROM cells WHERE tile = ? AND z > ?\", \r\n [old_tile_id, min_height]\r\n )\r\n \r\n if updated_tiles:\r\n db.execute_non_query(\r\n \"UPDATE cells SET tile = ? WHERE tile = ? AND z > ?\",\r\n [new_tile_id, old_tile_id, min_height]\r\n )\r\n \r\n pool.return_connection(db_path)\r\n```\r\n\r\n## Requirements\r\n\r\n- [duckdb](https://pypi.org/project/duckdb/) >= 0.8.0\r\n\r\n## Version History\r\n\r\n- **v1.0.0** - Initial release\r\n\r\n## License\r\n\r\nSee the main project for license information.\r\n",
"bugtrack_url": null,
"license": null,
"summary": "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.",
"version": "1.0",
"project_urls": {
"Homepage": "https://github.com/uaineteine/udbcore"
},
"split_keywords": [],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "79c4f7ed82cc22f54c843260ef8c89b9e901c06d48f940e50b7878f6d1c25f0b",
"md5": "f9492e5cab1d56d905c8d1b9446c53d8",
"sha256": "0bea43e013a67d21c010f83a5e6fe74bdf837f77ab3cb6f94b40612d59863023"
},
"downloads": -1,
"filename": "udbcore-1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "f9492e5cab1d56d905c8d1b9446c53d8",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 32975,
"upload_time": "2025-09-16T14:37:33",
"upload_time_iso_8601": "2025-09-16T14:37:33.082643Z",
"url": "https://files.pythonhosted.org/packages/79/c4/f7ed82cc22f54c843260ef8c89b9e901c06d48f940e50b7878f6d1c25f0b/udbcore-1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "8ac0051083c33eb5c2cbbe4c254b153b4f05957af1071d1dcebd9f826b8e029b",
"md5": "3786f8a98f4fa5b46dd0fe994b407d7e",
"sha256": "b62bc68d5a47f6616a9160d063b9b3f7fb5c69f592ee05052479cb35ef9e43c5"
},
"downloads": -1,
"filename": "udbcore-1.0.tar.gz",
"has_sig": false,
"md5_digest": "3786f8a98f4fa5b46dd0fe994b407d7e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 32838,
"upload_time": "2025-09-16T14:37:36",
"upload_time_iso_8601": "2025-09-16T14:37:36.802357Z",
"url": "https://files.pythonhosted.org/packages/8a/c0/051083c33eb5c2cbbe4c254b153b4f05957af1071d1dcebd9f826b8e029b/udbcore-1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-16 14:37:36",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "uaineteine",
"github_project": "udbcore",
"github_not_found": true,
"lcname": "udbcore"
}
