Skip to content

jmcore.models

jmcore.models

Core data models using Pydantic for validation and serialization.

Attributes

DIRECTORY_NODES_MAINNET: list[str] = ['satoshi2vcg5e2ept7tjkzlkpomkobqmgtsjzegg6wipnoajadissead.onion:5222', 'coinjointovy3eq5fjygdwpkbcdx63d7vd4g32mw7y553uj3kjjzkiqd.onion:5222', 'nakamotourflxwjnjpnrk7yc2nhkf6r62ed4gdfxmmn5f4saw5q5qoyd.onion:5222', 'odpwaf67rs5226uabcamvypg3y4bngzmfk7255flcdodesqhsvkptaid.onion:5222', 'jmarketxf5wc4aldf3slm5u6726zsky52bqnfv6qyxe5hnafgly6yuyd.onion:5222'] module-attribute

DIRECTORY_NODES_SIGNET: list[str] = ['signetvaxgd3ivj4tml4g6ed3samaa2rscre2gyeyohncmwk4fbesiqd.onion:5222'] module-attribute

DIRECTORY_NODES_TESTNET: list[str] = [] module-attribute

Classes

FidelityBond

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
class FidelityBond(BaseModel):
    counterparty: str
    utxo_txid: str = Field(..., pattern=r"^[0-9a-fA-F]{64}$")
    utxo_vout: int = Field(..., ge=0)
    bond_value: int | None = Field(default=None, ge=0)
    locktime: int = Field(..., ge=0)
    amount: int = Field(default=0, ge=0)
    script: str
    utxo_confirmations: int = Field(..., ge=0)
    utxo_confirmation_timestamp: int = Field(default=0, ge=0)
    cert_expiry: int = Field(..., ge=0)
    directory_node: str | None = None
    directory_nodes: list[str] = Field(
        default_factory=list,
        description="All directory nodes that announced this bond (for statistics)",
    )
    fidelity_bond_data: dict[str, Any] | None = None
Attributes
amount: int = Field(default=0, ge=0) class-attribute instance-attribute
bond_value: int | None = Field(default=None, ge=0) class-attribute instance-attribute
cert_expiry: int = Field(..., ge=0) class-attribute instance-attribute
counterparty: str instance-attribute
directory_node: str | None = None class-attribute instance-attribute
directory_nodes: list[str] = Field(default_factory=list, description='All directory nodes that announced this bond (for statistics)') class-attribute instance-attribute
fidelity_bond_data: dict[str, Any] | None = None class-attribute instance-attribute
locktime: int = Field(..., ge=0) class-attribute instance-attribute
script: str instance-attribute
utxo_confirmation_timestamp: int = Field(default=0, ge=0) class-attribute instance-attribute
utxo_confirmations: int = Field(..., ge=0) class-attribute instance-attribute
utxo_txid: str = Field(..., pattern='^[0-9a-fA-F]{64}$') class-attribute instance-attribute
utxo_vout: int = Field(..., ge=0) class-attribute instance-attribute

HandshakeRequest

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
182
183
184
185
186
187
188
189
class HandshakeRequest(BaseModel):
    app_name: str = "JoinMarket"
    directory: bool = False
    location_string: str
    proto_ver: int
    features: dict[str, Any] = Field(default_factory=dict)
    nick: str = Field(..., min_length=1)
    network: NetworkType
Attributes
app_name: str = 'JoinMarket' class-attribute instance-attribute
directory: bool = False class-attribute instance-attribute
features: dict[str, Any] = Field(default_factory=dict) class-attribute instance-attribute
location_string: str instance-attribute
network: NetworkType instance-attribute
nick: str = Field(..., min_length=1) class-attribute instance-attribute
proto_ver: int instance-attribute

HandshakeResponse

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
192
193
194
195
196
197
198
199
200
201
class HandshakeResponse(BaseModel):
    app_name: str = "JoinMarket"
    directory: bool = True
    proto_ver_min: int
    proto_ver_max: int
    features: dict[str, Any] = Field(default_factory=dict)
    accepted: bool
    nick: str = Field(..., min_length=1)
    network: NetworkType
    motd: str = "JoinMarket Directory Server"
Attributes
accepted: bool instance-attribute
app_name: str = 'JoinMarket' class-attribute instance-attribute
directory: bool = True class-attribute instance-attribute
features: dict[str, Any] = Field(default_factory=dict) class-attribute instance-attribute
motd: str = 'JoinMarket Directory Server' class-attribute instance-attribute
network: NetworkType instance-attribute
nick: str = Field(..., min_length=1) class-attribute instance-attribute
proto_ver_max: int instance-attribute
proto_ver_min: int instance-attribute

MessageEnvelope

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class MessageEnvelope(BaseModel):
    message_type: int = Field(..., ge=0)
    payload: str
    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))

    def to_bytes(self) -> bytes:
        import json

        result = json.dumps({"type": self.message_type, "line": self.payload}).encode("utf-8")
        return result

    @classmethod
    def from_bytes(
        cls, data: bytes, max_line_length: int = 65536, max_json_nesting_depth: int = 10
    ) -> MessageEnvelope:
        """
        Parse a message envelope from bytes with security limits.

        Args:
            data: Raw message bytes (without \\r\\n terminator)
            max_line_length: Maximum allowed line length in bytes (default 64KB)
            max_json_nesting_depth: Maximum JSON nesting depth (default 10)

        Returns:
            Parsed MessageEnvelope

        Raises:
            MessageParsingError: If message exceeds security limits
            json.JSONDecodeError: If JSON is malformed
        """
        import json

        # Check line length BEFORE parsing to prevent DoS
        if len(data) > max_line_length:
            raise MessageParsingError(
                f"Message line length {len(data)} exceeds maximum of {max_line_length} bytes"
            )

        # Parse JSON
        obj = json.loads(data)

        # Validate nesting depth BEFORE creating model
        validate_json_nesting_depth(obj, max_json_nesting_depth)

        return cls(message_type=obj["type"], payload=obj["line"])
Attributes
message_type: int = Field(..., ge=0) class-attribute instance-attribute
payload: str instance-attribute
timestamp: datetime = Field(default_factory=(lambda: datetime.now(UTC))) class-attribute instance-attribute
Functions
from_bytes(data: bytes, max_line_length: int = 65536, max_json_nesting_depth: int = 10) -> MessageEnvelope classmethod

Parse a message envelope from bytes with security limits.

Args: data: Raw message bytes (without \r\n terminator) max_line_length: Maximum allowed line length in bytes (default 64KB) max_json_nesting_depth: Maximum JSON nesting depth (default 10)

Returns: Parsed MessageEnvelope

Raises: MessageParsingError: If message exceeds security limits json.JSONDecodeError: If JSON is malformed

Source code in jmcore/src/jmcore/models.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
@classmethod
def from_bytes(
    cls, data: bytes, max_line_length: int = 65536, max_json_nesting_depth: int = 10
) -> MessageEnvelope:
    """
    Parse a message envelope from bytes with security limits.

    Args:
        data: Raw message bytes (without \\r\\n terminator)
        max_line_length: Maximum allowed line length in bytes (default 64KB)
        max_json_nesting_depth: Maximum JSON nesting depth (default 10)

    Returns:
        Parsed MessageEnvelope

    Raises:
        MessageParsingError: If message exceeds security limits
        json.JSONDecodeError: If JSON is malformed
    """
    import json

    # Check line length BEFORE parsing to prevent DoS
    if len(data) > max_line_length:
        raise MessageParsingError(
            f"Message line length {len(data)} exceeds maximum of {max_line_length} bytes"
        )

    # Parse JSON
    obj = json.loads(data)

    # Validate nesting depth BEFORE creating model
    validate_json_nesting_depth(obj, max_json_nesting_depth)

    return cls(message_type=obj["type"], payload=obj["line"])
to_bytes() -> bytes
Source code in jmcore/src/jmcore/models.py 
140
141
142
143
144
def to_bytes(self) -> bytes:
    import json

    result = json.dumps({"type": self.message_type, "line": self.payload}).encode("utf-8")
    return result

MessageParsingError

Bases: Exception

Exception raised when message parsing fails due to security limits.

Source code in jmcore/src/jmcore/models.py 
17
18
19
20
class MessageParsingError(Exception):
    """Exception raised when message parsing fails due to security limits."""

    pass

NetworkType

Bases: StrEnum

Source code in jmcore/src/jmcore/models.py 
83
84
85
86
87
class NetworkType(StrEnum):
    MAINNET = "mainnet"
    TESTNET = "testnet"
    SIGNET = "signet"
    REGTEST = "regtest"
Attributes
MAINNET = 'mainnet' class-attribute instance-attribute
REGTEST = 'regtest' class-attribute instance-attribute
SIGNET = 'signet' class-attribute instance-attribute
TESTNET = 'testnet' class-attribute instance-attribute

Offer

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
class Offer(BaseModel):
    counterparty: str = Field(..., min_length=1)
    oid: int = Field(..., ge=0)
    ordertype: OfferType
    minsize: int = Field(..., ge=0)
    maxsize: int = Field(..., ge=0)
    txfee: int = Field(..., ge=0)
    cjfee: str | int
    fidelity_bond_value: int = Field(default=0, ge=0)
    directory_node: str | None = None
    directory_nodes: list[str] = Field(
        default_factory=list,
        description="All directory nodes that announced this offer (for statistics)",
    )
    fidelity_bond_data: dict[str, Any] | None = None
    neutrino_compat: bool = Field(
        default=False,
        description="Maker requires extended UTXO format (neutrino-compatible backend)",
    )
    features: dict[str, bool] = Field(
        default_factory=dict,
        description="Features supported by this maker (from handshake)",
    )
    directly_reachable: bool | None = Field(
        default=None,
        description="Whether maker is directly reachable via their onion address (None = not checked)",
    )

    @field_validator("cjfee")
    @classmethod
    def validate_cjfee(cls, v: str | int, info) -> str | int:
        ordertype = info.data.get("ordertype")
        if ordertype in (OfferType.SW0_ABSOLUTE, OfferType.SWA_ABSOLUTE):
            return int(v)
        return str(v)

    def is_absolute_fee(self) -> bool:
        return is_absolute_offer_type(self.ordertype)

    def calculate_fee(self, amount: int) -> int:
        return calculate_cj_fee(self.ordertype, self.cjfee, amount)
Attributes
cjfee: str | int instance-attribute
counterparty: str = Field(..., min_length=1) class-attribute instance-attribute
directly_reachable: bool | None = Field(default=None, description='Whether maker is directly reachable via their onion address (None = not checked)') class-attribute instance-attribute
directory_node: str | None = None class-attribute instance-attribute
directory_nodes: list[str] = Field(default_factory=list, description='All directory nodes that announced this offer (for statistics)') class-attribute instance-attribute
features: dict[str, bool] = Field(default_factory=dict, description='Features supported by this maker (from handshake)') class-attribute instance-attribute
fidelity_bond_data: dict[str, Any] | None = None class-attribute instance-attribute
fidelity_bond_value: int = Field(default=0, ge=0) class-attribute instance-attribute
maxsize: int = Field(..., ge=0) class-attribute instance-attribute
minsize: int = Field(..., ge=0) class-attribute instance-attribute
neutrino_compat: bool = Field(default=False, description='Maker requires extended UTXO format (neutrino-compatible backend)') class-attribute instance-attribute
oid: int = Field(..., ge=0) class-attribute instance-attribute
ordertype: OfferType instance-attribute
txfee: int = Field(..., ge=0) class-attribute instance-attribute
Functions
calculate_fee(amount: int) -> int
Source code in jmcore/src/jmcore/models.py 
275
276
def calculate_fee(self, amount: int) -> int:
    return calculate_cj_fee(self.ordertype, self.cjfee, amount)
is_absolute_fee() -> bool
Source code in jmcore/src/jmcore/models.py 
272
273
def is_absolute_fee(self) -> bool:
    return is_absolute_offer_type(self.ordertype)
validate_cjfee(v: str | int, info) -> str | int classmethod
Source code in jmcore/src/jmcore/models.py 
264
265
266
267
268
269
270
@field_validator("cjfee")
@classmethod
def validate_cjfee(cls, v: str | int, info) -> str | int:
    ordertype = info.data.get("ordertype")
    if ordertype in (OfferType.SW0_ABSOLUTE, OfferType.SWA_ABSOLUTE):
        return int(v)
    return str(v)

OfferType

Bases: StrEnum

Source code in jmcore/src/jmcore/models.py 
204
205
206
207
208
class OfferType(StrEnum):
    SW0_ABSOLUTE = "sw0absoffer"
    SW0_RELATIVE = "sw0reloffer"
    SWA_ABSOLUTE = "swabsoffer"
    SWA_RELATIVE = "swreloffer"
Attributes
SW0_ABSOLUTE = 'sw0absoffer' class-attribute instance-attribute
SW0_RELATIVE = 'sw0reloffer' class-attribute instance-attribute
SWA_ABSOLUTE = 'swabsoffer' class-attribute instance-attribute
SWA_RELATIVE = 'swreloffer' class-attribute instance-attribute

OrderBook

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
class OrderBook(BaseModel):
    offers: list[Offer] = Field(default_factory=list)
    fidelity_bonds: list[FidelityBond] = Field(default_factory=list)
    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))
    directory_nodes: list[str] = Field(default_factory=list)

    def add_offers(self, offers: list[Offer], directory_node: str) -> None:
        for offer in offers:
            offer.directory_node = directory_node
        self.offers.extend(offers)
        if directory_node not in self.directory_nodes:
            self.directory_nodes.append(directory_node)

    def add_fidelity_bonds(self, bonds: list[FidelityBond], directory_node: str) -> None:
        for bond in bonds:
            bond.directory_node = directory_node
        self.fidelity_bonds.extend(bonds)

    def get_offers_by_directory(self) -> dict[str, list[Offer]]:
        """Get offers grouped by directory node.

        This uses the directory_nodes list (plural) which tracks all directories
        that announced each offer, so an offer will appear under multiple
        directories if it was announced by multiple directories.
        """
        result: dict[str, list[Offer]] = {}
        for offer in self.offers:
            # Use directory_nodes (plural) if populated, otherwise fallback to directory_node
            nodes = offer.directory_nodes if offer.directory_nodes else []
            if not nodes and offer.directory_node:
                nodes = [offer.directory_node]
            if not nodes:
                nodes = ["unknown"]
            for node in nodes:
                if node not in result:
                    result[node] = []
                result[node].append(offer)
        return result
Attributes
directory_nodes: list[str] = Field(default_factory=list) class-attribute instance-attribute
fidelity_bonds: list[FidelityBond] = Field(default_factory=list) class-attribute instance-attribute
offers: list[Offer] = Field(default_factory=list) class-attribute instance-attribute
timestamp: datetime = Field(default_factory=(lambda: datetime.now(UTC))) class-attribute instance-attribute
Functions
add_fidelity_bonds(bonds: list[FidelityBond], directory_node: str) -> None
Source code in jmcore/src/jmcore/models.py 
311
312
313
314
def add_fidelity_bonds(self, bonds: list[FidelityBond], directory_node: str) -> None:
    for bond in bonds:
        bond.directory_node = directory_node
    self.fidelity_bonds.extend(bonds)
add_offers(offers: list[Offer], directory_node: str) -> None
Source code in jmcore/src/jmcore/models.py 
304
305
306
307
308
309
def add_offers(self, offers: list[Offer], directory_node: str) -> None:
    for offer in offers:
        offer.directory_node = directory_node
    self.offers.extend(offers)
    if directory_node not in self.directory_nodes:
        self.directory_nodes.append(directory_node)
get_offers_by_directory() -> dict[str, list[Offer]]

Get offers grouped by directory node.

This uses the directory_nodes list (plural) which tracks all directories that announced each offer, so an offer will appear under multiple directories if it was announced by multiple directories.

Source code in jmcore/src/jmcore/models.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def get_offers_by_directory(self) -> dict[str, list[Offer]]:
    """Get offers grouped by directory node.

    This uses the directory_nodes list (plural) which tracks all directories
    that announced each offer, so an offer will appear under multiple
    directories if it was announced by multiple directories.
    """
    result: dict[str, list[Offer]] = {}
    for offer in self.offers:
        # Use directory_nodes (plural) if populated, otherwise fallback to directory_node
        nodes = offer.directory_nodes if offer.directory_nodes else []
        if not nodes and offer.directory_node:
            nodes = [offer.directory_node]
        if not nodes:
            nodes = ["unknown"]
        for node in nodes:
            if node not in result:
                result[node] = []
            result[node].append(offer)
    return result

PeerInfo

Bases: BaseModel

Source code in jmcore/src/jmcore/models.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class PeerInfo(BaseModel):
    nick: str = Field(..., min_length=1, max_length=64)
    onion_address: str = Field(..., pattern=r"^[a-z2-7]{56}\.onion$|^NOT-SERVING-ONION$")
    port: int = Field(..., ge=-1, le=65535)
    status: PeerStatus = PeerStatus.UNCONNECTED
    is_directory: bool = False
    network: NetworkType = NetworkType.MAINNET
    last_seen: datetime | None = None
    features: dict[str, Any] = Field(default_factory=dict)
    protocol_version: int = Field(default=5, ge=5, le=10)  # Negotiated protocol version
    neutrino_compat: bool = False  # True if peer supports extended UTXO metadata

    @field_validator("onion_address")
    @classmethod
    def validate_onion(cls, v: str) -> str:
        if v == "NOT-SERVING-ONION":
            return v
        if not v.endswith(".onion"):
            raise ValueError("Invalid onion address")
        return v

    @field_validator("port")
    @classmethod
    def validate_port(cls, v: int, info) -> int:
        if v == -1 and info.data.get("onion_address") == "NOT-SERVING-ONION":
            return v
        if v < 1 or v > 65535:
            raise ValueError("Port must be between 1 and 65535")
        return v

    @cached_property
    def location_string(self) -> str:
        if self.onion_address == "NOT-SERVING-ONION":
            return "NOT-SERVING-ONION"
        return f"{self.onion_address}:{self.port}"

    def supports_extended_utxo(self) -> bool:
        """Check if this peer supports extended UTXO format (neutrino_compat)."""
        # With feature-based detection, we check the neutrino_compat flag
        # which is set from the features dict during handshake
        return self.neutrino_compat

    model_config = {"frozen": False}
Attributes
features: dict[str, Any] = Field(default_factory=dict) class-attribute instance-attribute
is_directory: bool = False class-attribute instance-attribute
last_seen: datetime | None = None class-attribute instance-attribute
location_string: str cached property
model_config = {'frozen': False} class-attribute instance-attribute
network: NetworkType = NetworkType.MAINNET class-attribute instance-attribute
neutrino_compat: bool = False class-attribute instance-attribute
nick: str = Field(..., min_length=1, max_length=64) class-attribute instance-attribute
onion_address: str = Field(..., pattern='^[a-z2-7]{56}\\.onion$|^NOT-SERVING-ONION$') class-attribute instance-attribute
port: int = Field(..., ge=(-1), le=65535) class-attribute instance-attribute
protocol_version: int = Field(default=5, ge=5, le=10) class-attribute instance-attribute
status: PeerStatus = PeerStatus.UNCONNECTED class-attribute instance-attribute
Functions
supports_extended_utxo() -> bool

Check if this peer supports extended UTXO format (neutrino_compat).

Source code in jmcore/src/jmcore/models.py 
126
127
128
129
130
def supports_extended_utxo(self) -> bool:
    """Check if this peer supports extended UTXO format (neutrino_compat)."""
    # With feature-based detection, we check the neutrino_compat flag
    # which is set from the features dict during handshake
    return self.neutrino_compat
validate_onion(v: str) -> str classmethod
Source code in jmcore/src/jmcore/models.py 
102
103
104
105
106
107
108
109
@field_validator("onion_address")
@classmethod
def validate_onion(cls, v: str) -> str:
    if v == "NOT-SERVING-ONION":
        return v
    if not v.endswith(".onion"):
        raise ValueError("Invalid onion address")
    return v
validate_port(v: int, info) -> int classmethod
Source code in jmcore/src/jmcore/models.py 
111
112
113
114
115
116
117
118
@field_validator("port")
@classmethod
def validate_port(cls, v: int, info) -> int:
    if v == -1 and info.data.get("onion_address") == "NOT-SERVING-ONION":
        return v
    if v < 1 or v > 65535:
        raise ValueError("Port must be between 1 and 65535")
    return v

PeerStatus

Bases: StrEnum

Source code in jmcore/src/jmcore/models.py 
76
77
78
79
80
class PeerStatus(StrEnum):
    UNCONNECTED = "unconnected"
    CONNECTED = "connected"
    HANDSHAKED = "handshaked"
    DISCONNECTED = "disconnected"
Attributes
CONNECTED = 'connected' class-attribute instance-attribute
DISCONNECTED = 'disconnected' class-attribute instance-attribute
HANDSHAKED = 'handshaked' class-attribute instance-attribute
UNCONNECTED = 'unconnected' class-attribute instance-attribute

Functions

calculate_cj_fee(offer_type: OfferType, cjfee: str | int, amount: int) -> int

Calculate actual CoinJoin fee based on offer type.

This is the canonical fee calculation used by both makers and takers.

Args: offer_type: Absolute or relative offer type cjfee: Fee value (int for absolute, string decimal for relative) amount: CoinJoin amount in satoshis

Returns: Actual fee in satoshis

Source code in jmcore/src/jmcore/models.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def calculate_cj_fee(offer_type: OfferType, cjfee: str | int, amount: int) -> int:
    """
    Calculate actual CoinJoin fee based on offer type.

    This is the canonical fee calculation used by both makers and takers.

    Args:
        offer_type: Absolute or relative offer type
        cjfee: Fee value (int for absolute, string decimal for relative)
        amount: CoinJoin amount in satoshis

    Returns:
        Actual fee in satoshis
    """
    if is_absolute_offer_type(offer_type):
        return int(cjfee)
    else:
        return calculate_relative_fee(amount, str(cjfee))

get_default_directory_nodes(network: NetworkType) -> list[str]

Get default directory nodes for a given network.

Source code in jmcore/src/jmcore/models.py 
64
65
66
67
68
69
70
71
72
73
def get_default_directory_nodes(network: NetworkType) -> list[str]:
    """Get default directory nodes for a given network."""
    if network == NetworkType.MAINNET:
        return DIRECTORY_NODES_MAINNET.copy()
    elif network == NetworkType.SIGNET:
        return DIRECTORY_NODES_SIGNET.copy()
    elif network == NetworkType.TESTNET:
        return DIRECTORY_NODES_TESTNET.copy()
    # Regtest has no default directory nodes - must be configured
    return []

is_absolute_offer_type(offer_type: OfferType) -> bool

Check if an offer type uses absolute fees.

Source code in jmcore/src/jmcore/models.py 
211
212
213
def is_absolute_offer_type(offer_type: OfferType) -> bool:
    """Check if an offer type uses absolute fees."""
    return offer_type in (OfferType.SW0_ABSOLUTE, OfferType.SWA_ABSOLUTE)

validate_json_nesting_depth(obj: Any, max_depth: int = 10, current_depth: int = 0) -> None

Validate that a JSON object does not exceed maximum nesting depth.

Args: obj: The object to validate (dict, list, or primitive) max_depth: Maximum allowed nesting depth current_depth: Current depth in recursion

Raises: MessageParsingError: If nesting depth exceeds max_depth

Source code in jmcore/src/jmcore/models.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def validate_json_nesting_depth(obj: Any, max_depth: int = 10, current_depth: int = 0) -> None:
    """
    Validate that a JSON object does not exceed maximum nesting depth.

    Args:
        obj: The object to validate (dict, list, or primitive)
        max_depth: Maximum allowed nesting depth
        current_depth: Current depth in recursion

    Raises:
        MessageParsingError: If nesting depth exceeds max_depth
    """
    if current_depth > max_depth:
        raise MessageParsingError(f"JSON nesting depth exceeds maximum of {max_depth}")

    if isinstance(obj, dict):
        for value in obj.values():
            validate_json_nesting_depth(value, max_depth, current_depth + 1)
    elif isinstance(obj, list):
        for item in obj:
            validate_json_nesting_depth(item, max_depth, current_depth + 1)