Skip to main content
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()

Build docs developers (and LLMs) love