Python API¶

import fdb

# REQUIRED: Set API version before any other FDB calls
fdb.api_version(730)

# Open the default database
db = fdb.open()

# Simple transaction using decorator
@fdb.transactional
def hello_world(tr):
    tr[b'hello'] = b'world'
    return tr[b'hello']

result = hello_world(db)
print(result)  # b'world'

import fdb
fdb.api_version(730)  # FoundationDB 7.3

# Default cluster file location
db = fdb.open()

# Custom cluster file
db = fdb.open('/path/to/fdb.cluster')

# With event loop (for async)
db = fdb.open(event_model='asyncio')

@fdb.transactional
def transfer_funds(tr, from_acct, to_acct, amount):
    """Transfer funds between accounts atomically."""
    from_key = fdb.tuple.pack(('accounts', from_acct))
    to_key = fdb.tuple.pack(('accounts', to_acct))

    from_balance = int(tr[from_key] or b'0')
    to_balance = int(tr[to_key] or b'0')

    if from_balance < amount:
        raise ValueError("Insufficient funds")

    tr[from_key] = str(from_balance - amount).encode()
    tr[to_key] = str(to_balance + amount).encode()

# Pass database as first argument
transfer_funds(db, 'alice', 'bob', 100)

with db.transaction() as tr:
    tr[b'key1'] = b'value1'
    tr[b'key2'] = b'value2'
    # Commits automatically on exit

tr = db.create_transaction()
try:
    tr[b'key'] = b'value'
    tr.commit().wait()
except fdb.FDBError as e:
    tr.on_error(e).wait()
    # Retry logic here

@fdb.transactional
def get_value(tr, key):
    return tr[key]  # Returns None if not found

@fdb.transactional
def get_range(tr, start, end):
    """Read all key-value pairs in range."""
    return list(tr.get_range(start, end))

@fdb.transactional
def get_all_users(tr):
    """Read all users using tuple keys."""
    start = fdb.tuple.pack(('users',))
    end = fdb.tuple.pack(('users', None))  # None is max value

    users = []
    for key, value in tr.get_range(start, end):
        _, user_id = fdb.tuple.unpack(key)
        users.append((user_id, value.decode()))
    return users

@fdb.transactional
def get_with_options(tr):
    # Limit results
    result = tr.get_range(start, end, limit=100)

    # Reverse order
    result = tr.get_range(start, end, reverse=True)

    # Streaming mode for large ranges
    result = tr.get_range(start, end, streaming_mode=fdb.StreamingMode.want_all)

@fdb.transactional
def get_users_by_city(tr, city):
    """Fetch users using a city index with GetMappedRange (experimental)."""
    # Index: ('city_index', city, user_id) -> b''
    # Data:  ('users', user_id) -> user_data

    # Required: GetMappedRange only works with snapshot reads and
    # READ_YOUR_WRITES_DISABLE.
    tr.options.set_read_your_writes_disable()

    index_start = fdb.tuple.pack(('city_index', city))
    index_end = fdb.tuple.pack(('city_index', city, None))

    # Define mapper to fetch user data for each index entry
    mapper = fdb.tuple.pack(('users',)) + b'{K[2]}'

    results = tr.snapshot.get_mapped_range(
        index_start, index_end,
        mapper=mapper
    )

    return [(key, value) for key, value in results]

@fdb.transactional
def write_data(tr):
    # Set a value
    tr[b'key'] = b'value'

    # Clear a key
    del tr[b'key']

    # Clear a range
    tr.clear_range(b'prefix\x00', b'prefix\xff')

@fdb.transactional
def atomic_ops(tr, key):
    # Atomic add (for counters)
    tr.add(key, struct.pack('<q', 1))

    # Bitwise operations
    tr.bit_and(key, b'\xff\x00')
    tr.bit_or(key, b'\x00\xff')
    tr.bit_xor(key, b'\xaa\x55')

    # Min/max
    tr.min(key, struct.pack('<q', 10))
    tr.max(key, struct.pack('<q', 100))

    # Compare and clear
    tr.compare_and_clear(key, b'expected')

from fdb import tuple

# Pack tuples into keys
key = tuple.pack(('users', 'alice', 'profile'))
# Result: b'\x02users\x00\x02alice\x00\x02profile\x00'

# Unpack keys into tuples
values = tuple.unpack(key)
# Result: ('users', 'alice', 'profile')

# Supported types: str, bytes, int, float, bool, None, UUID
key = tuple.pack(('data', 42, 3.14, True, None))

# Open the directory layer
directory = fdb.directory.create_or_open(db, ('myapp',))

# Create subspaces
users = directory.create_or_open(db, ('users',))
orders = directory.create_or_open(db, ('orders',))

# Use subspace for keys
@fdb.transactional
def create_user(tr, user_id, name):
    tr[users.pack((user_id, 'name'))] = name.encode()
    tr[users.pack((user_id, 'created'))] = str(time.time()).encode()

@fdb.transactional
def safe_operation(tr):
    # Transactional decorator handles retries
    value = tr[b'counter']
    tr[b'counter'] = str(int(value or b'0') + 1).encode()

try:
    result = some_operation(db)
except fdb.FDBError as e:
    if e.code == 1007:  # past_version
        print("Transaction too old")
    elif e.code == 1009:  # future_version
        print("Cluster version ahead")
    else:
        raise

import asyncio
import fdb

fdb.api_version(730)
db = fdb.open()

async def async_operation():
    @fdb.transactional
    async def read_write(tr):
        value = await tr[b'key']
        tr[b'key2'] = value + b'_modified'
        return value

    return await read_write(db)

asyncio.run(async_operation())

import struct

@fdb.transactional
def increment(tr, counter_key):
    tr.add(counter_key, struct.pack('<q', 1))

@fdb.transactional
def get_count(tr, counter_key):
    value = tr[counter_key]
    if value is None:
        return 0
    return struct.unpack('<q', value)[0]

@fdb.transactional
def exists(tr, key):
    return tr[key] is not None

@fdb.transactional
def get_page(tr, prefix, page_size, last_key=None):
    start = last_key + b'\x00' if last_key else prefix
    end = fdb.KeySelector.first_greater_or_equal(prefix + b'\xff')

    results = list(tr.get_range(start, end, limit=page_size + 1))

    has_more = len(results) > page_size
    if has_more:
        results = results[:page_size]

    return results, has_more