Database

Core database operations for jasonisnthappy Python bindings

The Database class represents a jasonisnthappy database instance.

Opening a database

open

Opens a database at the specified path.

@staticmethod
def open(path: str) -> Database

path · str · required

Path to the database file

Database · Database

Database instance

Example:

from jasonisnthappy import Database

db = Database.open("./my_database.db")
try:
    # Use database
    pass
finally:
    db.close()

open_with_options

Opens a database with custom options.

@staticmethod
def open_with_options(path: str, options: CDatabaseOptions) -> Database

path · str · required

Path to the database file

options · CDatabaseOptions · required

Custom database configuration options

Database · Database

Database instance

Example:

opts = Database.default_database_options()
opts.read_only = True
opts.cache_size = 2000

db = Database.open_with_options("./my_database.db", opts)

default_database_options

Returns the default database options.

@staticmethod
def default_database_options() -> CDatabaseOptions

CDatabaseOptions · CDatabaseOptions

Default database configuration

Database management

close

Closes the database and frees associated resources.

def close() -> None

Example:

db.close()

Context manager support

Use the with statement for automatic resource cleanup.

Example:

with Database.open("./my_database.db") as db:
    # Database is automatically closed when exiting the block
    tx = db.begin_transaction()
    # ...

begin_transaction

Begins a new transaction.

def begin_transaction() -> Transaction

Transaction · Transaction

Transaction instance

Example:

tx = db.begin_transaction()
try:
    # Perform operations
    tx.commit()
except Exception:
    tx.rollback()
    raise

get_collection

Gets a collection reference for non-transactional operations.

def get_collection(name: str) -> Collection

name · str · required

Name of the collection

Collection · Collection

Collection handle

Example:

coll = db.get_collection("users")
try:
    doc_id = coll.insert({"name": "Alice", "age": 30})
finally:
    coll.close()

Configuration

set_transaction_config

Sets the transaction retry configuration.

def set_transaction_config(config: CTransactionConfig) -> None

config · CTransactionConfig · required

Transaction configuration with retry settings

Example:

config = Database.default_transaction_config()
config.max_retries = 5
config.retry_backoff_base_ms = 10
db.set_transaction_config(config)

get_transaction_config

Gets the current transaction configuration.

def get_transaction_config() -> CTransactionConfig

CTransactionConfig · CTransactionConfig

Current transaction configuration

set_auto_checkpoint_threshold

Sets the auto-checkpoint threshold in WAL frames.

def set_auto_checkpoint_threshold(threshold: int) -> None

threshold · int · required

Number of WAL frames before automatic checkpoint

Database information

get_path

Gets the database file path.

def get_path() -> str

str · str

Database file path

is_read_only

Checks if the database is read-only.

def is_read_only() -> bool

bool · bool

True if the database is read-only

list_collections

Lists all collections in the database.

def list_collections() -> List[str]

List[str] · List[str]

List of collection names

Example:

collections = db.list_collections()
for name in collections:
    print(f"Collection: {name}")

collection_stats

Gets statistics for a collection.

def collection_stats(collection_name: str) -> Dict[str, Any]

collection_name · str · required

Name of the collection

Dict[str, Any] · Dict[str, Any]

Collection statistics

database_info

Gets database information.

def database_info() -> Dict[str, Any]

Dict[str, Any] · Dict[str, Any]

Database information including path, size, collections, etc.

max_bulk_operations

Returns the maximum number of bulk operations allowed.

def max_bulk_operations() -> int

int · int

Maximum bulk operations limit

max_document_size

Returns the maximum document size in bytes.

def max_document_size() -> int

int · int

Maximum document size in bytes

max_request_body_size

Returns the maximum HTTP request body size in bytes.

def max_request_body_size() -> int

int · int

Maximum request body size in bytes

Index management

create_index

Creates a single-field index.

def create_index(collection_name: str, index_name: str, field: str, unique: bool = False) -> None

collection_name · str · required

Name of the collection

index_name · str · required

Name for the index

field · str · required

Field to index

unique · bool · default: "False"

Whether the index should enforce uniqueness

Example:

db.create_index("users", "email_idx", "email", unique=True)

create_compound_index

Creates a compound index on multiple fields.

def create_compound_index(collection_name: str, index_name: str, fields: List[str], unique: bool = False) -> None

collection_name · str · required

Name of the collection

index_name · str · required

Name for the index

fields · List[str] · required

Fields to include in the compound index

unique · bool · default: "False"

Whether the index should enforce uniqueness

Example:

db.create_compound_index("users", "name_age_idx", ["name", "age"])

create_text_index

Creates a full-text search index.

def create_text_index(collection_name: str, index_name: str, field: str) -> None

collection_name · str · required

Name of the collection

index_name · str · required

Name for the index

field · str · required

Text field to index

Example:

db.create_text_index("articles", "content_text_idx", "content")

drop_index

Drops an index from a collection.

def drop_index(collection_name: str, index_name: str) -> None

collection_name · str · required

Name of the collection

index_name · str · required

Name of the index to drop

Example:

db.drop_index("users", "email_idx")

list_indexes

Lists all indexes for a collection.

def list_indexes(collection_name: str) -> List[Dict[str, Any]]

collection_name · str · required

Name of the collection

List[Dict[str, Any]] · List[Dict[str, Any]]

List of index metadata

Example:

indexes = db.list_indexes("users")
for idx in indexes:
    print(f"Index: {idx['name']}, Fields: {idx['fields']}, Unique: {idx['unique']}")

Schema validation

set_schema

Sets a JSON schema for validation.

def set_schema(collection_name: str, schema: Dict[str, Any]) -> None

collection_name · str · required

Name of the collection

schema · Dict[str, Any] · required

JSON schema for validation

Example:

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "number"},
        "email": {"type": "string", "format": "email"},
    },
    "required": ["name", "email"],
}
db.set_schema("users", schema)

get_schema

Gets the JSON schema for a collection.

def get_schema(collection_name: str) -> Optional[Dict[str, Any]]

collection_name · str · required

Name of the collection

Optional[Dict[str, Any]] · Optional[Dict[str, Any]]

JSON schema, or None if no schema is set

remove_schema

Removes the JSON schema from a collection.

def remove_schema(collection_name: str) -> None

collection_name · str · required

Name of the collection

Maintenance

checkpoint

Performs a manual WAL checkpoint.

def checkpoint() -> None

backup

Creates a backup of the database.

def backup(dest_path: str) -> None

dest_path · str · required

Destination path for the backup file

Example:

db.backup("/backups/my_database_backup.db")

verify_backup

Verifies the integrity of a backup and returns backup info.

def verify_backup(backup_path: str) -> Dict[str, Any]

backup_path · str · required

Path to the backup file

Dict[str, Any] · Dict[str, Any]

Backup verification information

garbage_collect

Performs garbage collection and returns stats.

def garbage_collect() -> Dict[str, Any]

Dict[str, Any] · Dict[str, Any]

Garbage collection statistics

metrics

Gets database metrics.

def metrics() -> Dict[str, Any]

Dict[str, Any] · Dict[str, Any]

Database metrics including reads, writes, cache hits, etc.

Example:

metrics = db.metrics()
print(f"Total reads: {metrics['reads']}")
print(f"Total writes: {metrics['writes']}")

frame_count

Gets the number of WAL frames.

def frame_count() -> int

int · int

Number of WAL frames

Web UI

start_web_ui

Starts the web UI server at the given address.

def start_web_ui(addr: str) -> WebServer

addr · str · required

Address to bind the web server to (e.g., “127.0.0.1:8080”)

WebServer · WebServer

WebServer handle to control the server

Example:

server = db.start_web_ui("127.0.0.1:8080")
print("Web UI available at http://127.0.0.1:8080")
# ... later ...
server.stop()