#!/usr/bin/env python3 # Commingle API – add transactions example import base64 import json import os import urllib.request from cryptography.hazmat.primitives import hashes # python3 -m pip install cryptography from cryptography.hazmat.primitives.asymmetric.x25519 import ( X25519PrivateKey, X25519PublicKey, ) from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.hkdf import HKDF # Get your API key and public key from Commingle → Settings → API access API_KEY = "00000000-0000-0000-0000-000000000000_xxxxxxxxxxxxxxxxxxxxxxxxxxxx" PUBLIC_KEY = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # Your user ID is automatically derived from the API key USER_ID = API_KEY.split("_", 1)[1] # Your transactions to be added via the Commingle API YOUR_TRANSACTIONS = [ { "idempotency": "txn_20260928_001", "data": { "x": "e", # type, required, allowed values: `e` = expense, `i` = income "t": "Supermarket", # title, optional, max length: 255 characters "d": "2026-09-28", # date, required, format YYYY-MM-DD "s": 46800, # time, optional, seconds from midnight 0..86399 "c": "USD", # currency, required, ISO 4217 code "e": None, # event id (1), optional "w": "_", # wallet id (1), required "f": { # currency rates (2), required, empty dictionary `{}` if not needed "USDEUR": 0.8788 }, "b": { # category breakdown (1, 3), required "g": 0.65, # 65% in groceries "cs": 0.35 # 35% in cosmetics }, "p": { # participants (4, 5), required USER_ID: { "o": -1996, # -19.96 USD (expense) "p": -1996, # -19.96 USD (expense) } }, "i": "w" # chart inclusion (6), required } }, { "idempotency": "txn_20260901_001", "data": { "x": "i", # type, required, allowed values: e = expense, i = income "t": "Salary", # title, optional, max length: 255 characters "d": "2026-09-01", # date, required, format YYYY-MM-DD "s": 46800, # time, optional, seconds from midnight 0..86399 "c": "EUR", # currency, required, ISO 4217 code "e": None, # event id (1), optional "w": "_", # wallet id (1), required, enter "_" for built-in Commingle wallet "f": {}, # currency rates (2), required, empty dictionary `{}` if not needed "b": { # category breakdown (1, 3), required "w": 1, # 100% in work }, "p": { # participants (4), required USER_ID: { "o": 1234567, # +12345.67 EUR (income) "p": 1234567, # +12345.67 EUR (income) } }, "i": "m" # chart inclusion (6), required } }, ] # (1) Find Event, Wallet and Category IDs at the bottom of their form screens. # Use "_" for the built-in Commingle wallet or Uncategorised category. # # (2) Provide all currency rates needed for conversions. # Example: the transaction is in USD, your primary currency is EUR, # the transaction belongs to an HKD event and a GBP wallet. # In this case, provide USDEUR, USDHKD and USDGBP. # If a required rate is missing, Commingle will offer to repair the data. # # (3) Category breakdown must sum to 1.0 for each transaction. # # (4) You can only include yourself as a participant. # Adding other participants will not create or recalculate debts. # # (5) `o` and `p` must be equal. # Use negative values for expenses and positive values for income. # Amounts use the currency's smallest unit, e.g. 1 USD = 100, # while 1 BHD = 1000. # # (6) `y` = yearly only # `q` = yearly and quarterly # `m` = yearly, quarterly and monthly # `w` = yearly, quarterly, monthly and weekly # `n` = never _TRADE_TYPES = {"e", "i"} _CHART_INCLUSION = {"n", "y", "q", "m", "w"} _CONTENT_KEY_INFO = b"trade-content-key-v1" def _b64url_encode(raw: bytes) -> str: return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") def _b64url_decode(value: str) -> bytes: padded = value + "=" * (-len(value) % 4) return base64.urlsafe_b64decode(padded) def _empty_to_null(value): if value == "": return None return value def normalize_trade(data: dict, user_id: str) -> dict: """Plaintext sealed into encryptedData. Nulls stay; weekly chart inclusion is omitted.""" trade_type = data.get("x") if trade_type not in _TRADE_TYPES: raise ValueError("data['x'] must be 'e' (expense) or 'i' (income).") if not isinstance(data.get("d"), str) or not data["d"]: raise ValueError("data['d'] is required (YYYY-MM-DD).") if not isinstance(data.get("c"), str) or not data["c"]: raise ValueError("data['c'] is required (ISO 4217).") if "w" not in data: raise ValueError("data['w'] is required ('_' is the built-in wallet).") seconds = data.get("s") if seconds is None: seconds = 0 seconds = int(seconds) if not 0 <= seconds <= 86399: raise ValueError("data['s'] must be seconds from midnight, 0..86399.") participants = data.get("p") if not isinstance(participants, dict) or list(participants) != [user_id]: raise ValueError("data['p'] must contain only your user id.") share = participants[user_id] if not isinstance(share, dict) or "o" not in share or "p" not in share: raise ValueError("data['p'][USER_ID] must include 'o' and 'p'.") forex = data.get("f") if forex is None: forex = {} if not isinstance(forex, dict): raise ValueError("data['f'] must be an object.") breakdown = data.get("b") if breakdown is None: breakdown = {} if not isinstance(breakdown, dict): raise ValueError("data['b'] must be an object.") plain = { "x": trade_type, "d": data["d"], "s": seconds, "t": _empty_to_null(data.get("t")), "c": data["c"], "si": _empty_to_null(data.get("si")), "e": _empty_to_null(data.get("e")), "w": _empty_to_null(data.get("w")), "w2": None, "f": forex, "b": breakdown, "p": participants, } inclusion = data.get("i") if inclusion is not None: if inclusion not in _CHART_INCLUSION: raise ValueError("data['i'] must be n, y, q, m or w.") # Weekly is the app default, so the key is left out. if inclusion != "w": plain["i"] = inclusion if data.get("j") is True: plain["j"] = True return plain def encrypt_trade(plaintext: dict, public_key: bytes) -> dict: """Solo trade envelope. Fresh content key and ephemeral key each call.""" content_key = os.urandom(32) ephemeral = X25519PrivateKey.generate() shared = ephemeral.exchange(X25519PublicKey.from_public_bytes(public_key)) wrap_key = HKDF( algorithm=hashes.SHA256(), length=32, salt=None, info=_CONTENT_KEY_INFO, ).derive(shared) wrapped = AESGCM(wrap_key).encrypt(b"\x00" * 12, content_key, None) plain_bytes = json.dumps( plaintext, separators=(",", ":"), ensure_ascii=False ).encode("utf-8") nonce = os.urandom(12) ciphertext = AESGCM(content_key).encrypt(nonce, plain_bytes, None) return { "encryptedData": _b64url_encode(ciphertext), "encryptedKey": { "ephemeralPublicKey": _b64url_encode( ephemeral.public_key().public_bytes_raw() ), "wrappedContentKey": _b64url_encode(wrapped), }, "nonce": _b64url_encode(nonce), } def build_payloads(transactions: list, public_key: str, user_id: str) -> list: recipient = _b64url_decode(public_key) payloads = [] for tx in transactions: if not isinstance(tx, dict) or not isinstance(tx.get("data"), dict): raise ValueError("Each transaction needs a data object.") envelope = encrypt_trade(normalize_trade(tx["data"], user_id), recipient) item = { "encryptedData": envelope["encryptedData"], "encryptedKey": envelope["encryptedKey"], "nonce": envelope["nonce"], } idempotency = tx.get("idempotency") if idempotency: item = {"idempotency": idempotency, **item} payloads.append(item) return payloads def send_transactions(payloads: list) -> str: body = json.dumps(payloads, separators=(",", ":"), ensure_ascii=False).encode( "utf-8" ) request = urllib.request.Request( "https://api.commingle.app/v1/transactions", data=body, method="POST", headers={ "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", }, ) with urllib.request.urlopen(request) as response: return response.read().decode() def main() -> None: payloads = build_payloads(YOUR_TRANSACTIONS, PUBLIC_KEY, USER_ID) print(json.dumps(payloads, indent=2, ensure_ascii=False)) print(send_transactions(payloads)) if __name__ == "__main__": main()