FoundationDB

# Python example (similar patterns in all languages)
import fdb

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

@fdb.transactional
def update_account_balance(tr, account_id, amount):
    balance_key = f"account:{account_id}:balance"
    current_balance = tr.get(balance_key).wait()
    if current_balance is None:
        current_balance = 0
    else:
        current_balance = int(current_balance)
    
    new_balance = current_balance + amount
    tr.set(balance_key, str(new_balance).encode())
    return new_balance

# Automatically retried on conflicts
new_balance = update_account_balance(db, "user123", 100)

// Network management
fdb_error_t fdb_select_api_version(int version);
fdb_error_t fdb_setup_network();
fdb_error_t fdb_run_network();
fdb_error_t fdb_stop_network();

// Database operations
fdb_error_t fdb_create_database(const char* cluster_file_path, FDBDatabase** out_database);
fdb_error_t fdb_database_create_transaction(FDBDatabase* database, FDBTransaction** out_transaction);

// Transaction operations
FDBFuture* fdb_transaction_get(FDBTransaction* tr, uint8_t const* key_name, int key_name_length, fdb_bool_t snapshot);
void fdb_transaction_set(FDBTransaction* tr, uint8_t const* key_name, int key_name_length, uint8_t const* value, int value_length);
FDBFuture* fdb_transaction_commit(FDBTransaction* tr);

def api_version(version: int) -> None: ...
def open(cluster_file: Optional[str] = None) -> Database: ...

@transactional
def transaction_function(tr: Transaction) -> Any: ...

class Database:
    def create_transaction(self) -> Transaction: ...
    def open_tenant(self, tenant_name: bytes) -> Tenant: ...

class Transaction:
    def get(self, key: KeyConvertible, snapshot: bool = False) -> Future: ...
    def get_range(self, begin: KeyConvertible, end: KeyConvertible, limit: int = 0) -> Iterable[KeyValue]: ...
    def set(self, key: KeyConvertible, value: ValueConvertible) -> None: ...
    def clear(self, key: KeyConvertible) -> None: ...
    def commit(self) -> Future: ...

public class FDB {
    public static FDB selectAPIVersion(int version);
    public Database open();
    public Database open(String clusterFilePath);
}

public interface Database {
    Transaction createTransaction();
    <T> T run(Function<Transaction, T> retryable);
    <T> T read(Function<ReadTransaction, T> retryable);
    Tenant openTenant(byte[] tenantName);
}

public interface Transaction extends ReadTransaction {
    CompletableFuture<byte[]> get(byte[] key);
    AsyncIterable<KeyValue> getRange(KeySelector begin, KeySelector end);
    void set(byte[] key, byte[] value);
    void clear(byte[] key);
    CompletableFuture<Void> commit();
}

func APIVersion(version int) error
func OpenDatabase(clusterFile string) (Database, error)
func StartNetwork() error
func StopNetwork() error

type Database interface {
    CreateTransaction() (Transaction, error)
    Transact(func(Transaction) (interface{}, error)) (interface{}, error)
    ReadTransact(func(ReadTransaction) (interface{}, error)) (interface{}, error)
    OpenTenant(tenantName []byte) (Tenant, error)
}

type Transaction interface {
    Get(key KeyConvertible) Future
    GetRange(r Range, options RangeOptions) RangeResult
    Set(key, value KeyConvertible)
    Clear(key KeyConvertible)
    Commit() Future
}

module FDB
  def self.api_version(version)
  def self.open(cluster_file = nil)
  def self.transactional(&block)

  class Database
    def create_transaction
    def transact(&block)
    def open_tenant(tenant_name)
  end

  class Transaction
    def get(key, snapshot: false)
    def get_range(begin_key, end_key, options = {})
    def []=(key, value)  # Alias for set
    def set(key, value)
    def clear(key)
    def commit
  end
end

# Tenant operations (similar across all languages)
tenant = db.open_tenant(b"app1")
tenant_tr = tenant.create_transaction()
# All operations scoped to tenant keyspace

import fdb.tuple as tuple

# Tuple encoding for structured keys
key = tuple.pack(("users", user_id, "profile"))
value = tuple.unpack(stored_key)

# Subspace for key prefixing
user_space = fdb.Subspace(("users",))
profile_key = user_space.pack((user_id, "profile"))

// Administrative blob operations (database level)
FDBFuture* fdb_database_blobbify_range(FDBDatabase* db, uint8_t const* begin_key, int begin_key_length,
                                       uint8_t const* end_key, int end_key_length);
FDBFuture* fdb_database_purge_blob_granules(FDBDatabase* db, uint8_t const* begin_key, int begin_key_length,
                                            uint8_t const* end_key, int end_key_length, 
                                            int64_t purge_version, fdb_bool_t force);

// Blob granule reading (transaction level)  
FDBResult* fdb_transaction_read_blob_granules(FDBTransaction* tr, uint8_t const* begin_key, int begin_key_length,
                                              uint8_t const* end_key, int end_key_length,
                                              int64_t beginVersion, int64_t readVersion,
                                              FDBReadBlobGranuleContext granuleContext);

# Python error predicates (fdb.predicates module)
def retryable(error: Exception) -> bool:
    """Check if error indicates transaction should be retried."""

def maybe_committed(error: Exception) -> bool:
    """Check if transaction may have been committed despite error."""

def retryable_not_committed(error: Exception) -> bool:
    """Check if error is retryable and transaction was not committed."""

// C API error predicates
fdb_bool_t fdb_error_predicate(int predicate_test, fdb_error_t code);

// Predicate test constants
#define FDB_ERROR_PREDICATE_RETRYABLE            50000
#define FDB_ERROR_PREDICATE_MAYBE_COMMITTED      50001
#define FDB_ERROR_PREDICATE_RETRYABLE_NOT_COMMITTED 50002

# Python automatic retry pattern
@fdb.transactional
def safe_operation(tr, key, value):
    # Automatically retries on retryable errors
    tr.set(key, value)
    return tr.get(key).wait()

# Manual error handling
def manual_retry_operation(db, key, value):
    tr = db.create_transaction() 
    while True:
        try:
            tr.set(key, value)
            tr.commit().wait()
            break
        except fdb.FDBError as e:
            if fdb.predicates.retryable(e):
                tr.on_error(e).wait()  # Wait for retry delay
                continue
            else:
                raise  # Non-retryable error

# Common types across bindings (Python syntax shown)
KeyConvertible = Union[bytes, str, tuple]
ValueConvertible = Union[bytes, str]

class KeyValue:
    key: bytes
    value: bytes

class KeySelector:
    key: bytes
    or_equal: bool
    offset: int

class Future:
    def wait(self) -> Any: ...
    def is_ready(self) -> bool: ...

class FDBError(Exception):
    code: int
    description: str