Skip to content

API reference

Auto-generated from the SDK's Google-style docstrings via mkdocstrings.

Client

zotniq.Zotniq

Zotniq SDK client for runtime DLP decisions.

Parameters:

Name Type Description Default
api_key Optional[str]

Zotniq API key (zot_sk_...). Reads ZOTNIQ_API_KEY env var when omitted. When neither is set, the client operates in local-only mode (regex detection, no network calls).

 None
base_url Optional[str]

Zotniq API base URL. Defaults to the public endpoint. Point at a self-hosted deployment for air-gapped scenarios.

 None
timeout Optional[float]

Per-request timeout in seconds. Default 30.

 None
max_retries Optional[int]

Number of retries on 429/5xx before raising. Default 2.

 None
on_decision Optional[Callable[..., Any]]

Optional callback fired after every preflight.check() returns, for SIEM forwarding. See zotniq.siem for built-in forwarders. Wired in commit 10.

 None
Example

client = Zotniq() result = client.preflight.check("hello", destination="AI_TOOL", mode="local") result.decision

Source code in zotniq/_client.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 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
133
134
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
180
181
class Zotniq:
    """Zotniq SDK client for runtime DLP decisions.

    Args:
        api_key: Zotniq API key (``zot_sk_...``). Reads ``ZOTNIQ_API_KEY``
            env var when omitted. When neither is set, the client operates
            in local-only mode (regex detection, no network calls).
        base_url: Zotniq API base URL. Defaults to the public endpoint.
            Point at a self-hosted deployment for air-gapped scenarios.
        timeout: Per-request timeout in seconds. Default 30.
        max_retries: Number of retries on 429/5xx before raising. Default 2.
        on_decision: Optional callback fired after every
            ``preflight.check()`` returns, for SIEM forwarding. See
            ``zotniq.siem`` for built-in forwarders. Wired in commit 10.

    Example:
        >>> client = Zotniq()
        >>> result = client.preflight.check("hello", destination="AI_TOOL", mode="local")
        >>> result.decision
        <Decision.ALLOWED: 'ALLOWED'>
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: Optional[float] = None,
        max_retries: Optional[int] = None,
        on_decision: Optional[Callable[..., Any]] = None,
    ) -> None:
        self._config = ClientConfig.resolve(
            api_key=api_key,
            base_url=base_url,
            timeout=timeout,
            max_retries=max_retries,
            on_decision=on_decision,
        )
        self._transport = HTTPTransport(self._config)
        self._hooks = HookDispatcher(self._config.on_decision)
        self.preflight = PreflightResource(
            self._transport,
            self._config.api_key,
            hooks=self._hooks,
            api_key_fingerprint=self._config.api_key_fingerprint,
        )

    def siem_stats(self) -> dict:
        """Return SIEM forwarder counters: delivered, dropped, queued."""
        return self._hooks.stats()

    def detect(self, text: str) -> list[Finding]:
        """Run local regex + validator detection. Zero network calls.

        Convenience over ``client.preflight.check(..., mode="local")`` when
        the caller only wants findings without a decision or masked_text.

        Example::

            client = Zotniq()
            [f.type for f in client.detect("[email protected]")]  # -> ["EMAIL"]
        """
        items = _detect_local(text)
        return [
            Finding(
                type=item.type.value if hasattr(item.type, "value") else str(item.type),
                count=item.count,
                sample=item.sample,
            )
            for item in items
        ]

    def mask(self, text: str) -> str:
        """Return the format-preserving masked version of ``text``.

        Always local. See ``zotniq.detection.mask_text`` for the same
        function without the client wrapper.

        Example:
            >>> Zotniq().mask("SSN 123-45-6789")
            'SSN XXX-XX-6789'
        """
        return _mask_local(text)

    @property
    def api_key(self) -> Optional[str]:
        return self._config.api_key

    @property
    def base_url(self) -> str:
        return self._config.base_url

    @property
    def timeout(self) -> float:
        return self._config.timeout

    @property
    def max_retries(self) -> int:
        return self._config.max_retries

    def with_options(
        self,
        timeout: Optional[float] = None,
        max_retries: Optional[int] = None,
    ) -> Zotniq:
        """Return a new client with per-call overrides.

        Matches the OpenAI SDK's ``with_options`` pattern. Cheap: reuses
        config, creates a new transport. Original client unchanged.

            long = client.with_options(timeout=120)
            result = long.preflight.check(text=very_large_text, destination="AI_TOOL")
        """
        new_config = self._config.with_overrides(
            timeout=timeout if timeout is not None else self._config.timeout,
            max_retries=max_retries if max_retries is not None else self._config.max_retries,
        )
        return Zotniq(
            api_key=new_config.api_key,
            base_url=new_config.base_url,
            timeout=new_config.timeout,
            max_retries=new_config.max_retries,
            on_decision=new_config.on_decision,
        )

    def close(self) -> None:
        """Release resources: SIEM queue flush + httpx connection pool.

        Safe to call multiple times. Prefer the context manager form
        (``with Zotniq(...) as client:``) for automatic cleanup.
        """
        self._hooks.flush(timeout=5.0)
        self._hooks.shutdown()
        self._transport.close()

    def __enter__(self) -> Zotniq:
        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        self.close()

    def __repr__(self) -> str:
        return (
            f"Zotniq(api_key={self._config.api_key_fingerprint}, "
            f"base_url={self._config.base_url!r})"
        )

siem_stats()

Return SIEM forwarder counters: delivered, dropped, queued.

Source code in zotniq/_client.py 
78
79
80
def siem_stats(self) -> dict:
    """Return SIEM forwarder counters: delivered, dropped, queued."""
    return self._hooks.stats()

detect(text)

Run local regex + validator detection. Zero network calls.

Convenience over client.preflight.check(..., mode="local") when the caller only wants findings without a decision or masked_text.

Example::

client = Zotniq()
[f.type for f in client.detect("[email protected]")]  # -> ["EMAIL"]
Source code in zotniq/_client.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def detect(self, text: str) -> list[Finding]:
    """Run local regex + validator detection. Zero network calls.

    Convenience over ``client.preflight.check(..., mode="local")`` when
    the caller only wants findings without a decision or masked_text.

    Example::

        client = Zotniq()
        [f.type for f in client.detect("[email protected]")]  # -> ["EMAIL"]
    """
    items = _detect_local(text)
    return [
        Finding(
            type=item.type.value if hasattr(item.type, "value") else str(item.type),
            count=item.count,
            sample=item.sample,
        )
        for item in items
    ]

mask(text)

Return the format-preserving masked version of text.

Always local. See zotniq.detection.mask_text for the same function without the client wrapper.

Example

Zotniq().mask("SSN 123-45-6789") 'SSN XXX-XX-6789'

Source code in zotniq/_client.py 
103
104
105
106
107
108
109
110
111
112
113
def mask(self, text: str) -> str:
    """Return the format-preserving masked version of ``text``.

    Always local. See ``zotniq.detection.mask_text`` for the same
    function without the client wrapper.

    Example:
        >>> Zotniq().mask("SSN 123-45-6789")
        'SSN XXX-XX-6789'
    """
    return _mask_local(text)

with_options(timeout=None, max_retries=None)

Return a new client with per-call overrides.

Matches the OpenAI SDK's with_options pattern. Cheap: reuses config, creates a new transport. Original client unchanged.

long = client.with_options(timeout=120)
result = long.preflight.check(text=very_large_text, destination="AI_TOOL")
Source code in zotniq/_client.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def with_options(
    self,
    timeout: Optional[float] = None,
    max_retries: Optional[int] = None,
) -> Zotniq:
    """Return a new client with per-call overrides.

    Matches the OpenAI SDK's ``with_options`` pattern. Cheap: reuses
    config, creates a new transport. Original client unchanged.

        long = client.with_options(timeout=120)
        result = long.preflight.check(text=very_large_text, destination="AI_TOOL")
    """
    new_config = self._config.with_overrides(
        timeout=timeout if timeout is not None else self._config.timeout,
        max_retries=max_retries if max_retries is not None else self._config.max_retries,
    )
    return Zotniq(
        api_key=new_config.api_key,
        base_url=new_config.base_url,
        timeout=new_config.timeout,
        max_retries=new_config.max_retries,
        on_decision=new_config.on_decision,
    )

close()

Release resources: SIEM queue flush + httpx connection pool.

Safe to call multiple times. Prefer the context manager form (with Zotniq(...) as client:) for automatic cleanup.

Source code in zotniq/_client.py 
156
157
158
159
160
161
162
163
164
def close(self) -> None:
    """Release resources: SIEM queue flush + httpx connection pool.

    Safe to call multiple times. Prefer the context manager form
    (``with Zotniq(...) as client:``) for automatic cleanup.
    """
    self._hooks.flush(timeout=5.0)
    self._hooks.shutdown()
    self._transport.close()

zotniq.AsyncZotniq

Async twin of Zotniq. See Zotniq for full argument docs.

Source code in zotniq/_async_client.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class AsyncZotniq:
    """Async twin of ``Zotniq``. See ``Zotniq`` for full argument docs."""

    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: Optional[float] = None,
        max_retries: Optional[int] = None,
        on_decision: Optional[Callable[..., Any]] = None,
    ) -> None:
        self._config = ClientConfig.resolve(
            api_key=api_key,
            base_url=base_url,
            timeout=timeout,
            max_retries=max_retries,
            on_decision=on_decision,
        )
        self._transport = AsyncHTTPTransport(self._config)
        self._hooks = HookDispatcher(self._config.on_decision)
        self.preflight = AsyncPreflightResource(
            self._transport,
            self._config.api_key,
            hooks=self._hooks,
            api_key_fingerprint=self._config.api_key_fingerprint,
        )

    @property
    def api_key(self) -> Optional[str]:
        return self._config.api_key

    @property
    def base_url(self) -> str:
        return self._config.base_url

    def detect(self, text: str) -> list[Finding]:
        """Local regex detection — synchronous, no network. Same as Zotniq.detect."""
        items = _detect_local(text)
        return [
            Finding(
                type=item.type.value if hasattr(item.type, "value") else str(item.type),
                count=item.count,
                sample=item.sample,
            )
            for item in items
        ]

    def mask(self, text: str) -> str:
        """Local mask — synchronous, no network. Same as Zotniq.mask."""
        return _mask_local(text)

    def siem_stats(self) -> dict:
        return self._hooks.stats()

    async def aclose(self) -> None:
        """Async cleanup: flush SIEM queue + close httpx AsyncClient pool."""
        self._hooks.flush(timeout=5.0)
        self._hooks.shutdown()
        await self._transport.aclose()

    async def __aenter__(self) -> AsyncZotniq:
        return self

    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        await self.aclose()

    def __repr__(self) -> str:
        return (
            f"AsyncZotniq(api_key={self._config.api_key_fingerprint}, "
            f"base_url={self._config.base_url!r})"
        )

detect(text)

Local regex detection — synchronous, no network. Same as Zotniq.detect.

Source code in zotniq/_async_client.py 
66
67
68
69
70
71
72
73
74
75
76
def detect(self, text: str) -> list[Finding]:
    """Local regex detection — synchronous, no network. Same as Zotniq.detect."""
    items = _detect_local(text)
    return [
        Finding(
            type=item.type.value if hasattr(item.type, "value") else str(item.type),
            count=item.count,
            sample=item.sample,
        )
        for item in items
    ]

mask(text)

Local mask — synchronous, no network. Same as Zotniq.mask.

Source code in zotniq/_async_client.py 
78
79
80
def mask(self, text: str) -> str:
    """Local mask — synchronous, no network. Same as Zotniq.mask."""
    return _mask_local(text)

aclose() async

Async cleanup: flush SIEM queue + close httpx AsyncClient pool.

Source code in zotniq/_async_client.py 
85
86
87
88
89
async def aclose(self) -> None:
    """Async cleanup: flush SIEM queue + close httpx AsyncClient pool."""
    self._hooks.flush(timeout=5.0)
    self._hooks.shutdown()
    await self._transport.aclose()

Preflight

zotniq.resources.preflight.PreflightResource

Attached at client.preflight. Owns preflight decision calls.

Source code in zotniq/resources/preflight.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 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
133
134
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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class PreflightResource:
    """Attached at ``client.preflight``. Owns preflight decision calls."""

    def __init__(
        self,
        transport: HTTPTransport,
        api_key: Optional[str],
        hooks: Optional[HookDispatcher] = None,
        api_key_fingerprint: str = "(no key)",
    ) -> None:
        self._transport = transport
        self._api_key = api_key
        self._hooks = hooks
        self._api_key_fingerprint = api_key_fingerprint

    def check(
        self,
        text: str,
        destination: DestinationLike,
        mode: Mode = "auto",
        actor_id: Optional[str] = None,
    ) -> PreflightResult:
        """Run a preflight check and return the decision.

        Args:
            text: The candidate content. Never persisted server-side under
                any retention_mode (Bug F invariant); only decision metadata
                lands in ``preflight_audit``.
            destination: Where the content is going. Drives which policy
                applies. Accepts ``Destination`` enum or a raw string.
            mode: See module docstring for resolution.
            actor_id: Who inside your app made this call. Populates the
                ``actor_id`` column on the server-side audit row so
                dashboard filters ("show me every decision for
                [email protected]") work end-to-end. Common values: end-user
                email, internal user id, service account label. Local-mode
                calls carry it on the response for parity but never leave
                the process.

        Returns:
            A ``PreflightResult`` with decision, findings, masked_text
            (when decision is ``ALLOWED_WITH_MASKING``), and a request_id
            for correlation with server-side audit rows.

        Raises:
            AuthError: mode="cloud" without an api_key configured.
            NetworkError: cloud call failed and no fallback was requested.
            ValidationError: server rejected the request body (rare).

        Example:
            >>> from zotniq import Zotniq
            >>> client = Zotniq()
            >>> result = client.preflight.check(
            ...     "email me at [email protected]",
            ...     destination="AI_TOOL",
            ...     mode="local",
            ... )
            >>> result.decision.value
            'ALLOWED_WITH_MASKING'
        """
        dest_str = destination.value if isinstance(destination, Destination) else str(destination)
        resolved_mode = self._resolve_mode(mode)

        if resolved_mode == "local":
            result = self._run_local(text=text, destination=dest_str)
        elif resolved_mode == "cloud":
            result = self._run_cloud(text=text, destination=dest_str, actor_id=actor_id)
        else:
            # cloud_with_fallback — attempt cloud, fall through to local on NetworkError
            try:
                result = self._run_cloud(text=text, destination=dest_str, actor_id=actor_id)
            except NetworkError as e:
                logger.warning("cloud call failed, falling back to local: %s", e)
                result = self._run_local(text=text, destination=dest_str)

        self._fire_hook(text=text, destination=dest_str, result=result)
        return result

    def _fire_hook(self, *, text: str, destination: str, result: PreflightResult) -> None:
        """Enqueue an on_decision event. No-op when no hook is configured."""
        if self._hooks is None:
            return
        ctx = build_context(
            text=text,
            destination=destination,
            result=result,
            api_key_fingerprint=self._api_key_fingerprint,
        )
        self._hooks.emit(_HookEvent(result=result, context=ctx))

    def _resolve_mode(self, mode: Mode) -> Mode:
        """Collapse ``auto`` to ``local`` or ``cloud`` based on key presence.

        Raises ``AuthError`` early when the caller explicitly asked for
        cloud without a key — better than sending an unauthenticated
        request the server will reject.
        """
        if mode == "auto":
            return "cloud" if self._api_key else "local"

        if mode in ("cloud", "cloud_with_fallback") and not self._api_key:
            raise AuthError(
                "Missing api_key. Set the ZOTNIQ_API_KEY environment variable or "
                "pass api_key= to Zotniq(). Get your team's API key at "
                "https://app.zotniq.ai/settings/api-keys. "
                "See https://docs.zotniq.ai/sdk/quickstart#authentication."
            )

        if mode not in ("auto", "local", "cloud", "cloud_with_fallback"):
            raise ValueError(
                f"invalid mode={mode!r}. Expected one of: auto, local, cloud, cloud_with_fallback."
            )

        return mode

    def _run_local(self, *, text: str, destination: str) -> PreflightResult:
        """Run regex + validator pass against the shared detection module.

        Decision logic mirrors what the server's classic evaluator would
        return for the same input under a default policy: PHI blocks
        outright, PII masks, clean passes.
        """
        detected: list[DetectedItem] = _detect_local(text)
        findings = [
            Finding(type=_type_str(item.type), count=item.count, sample=item.sample)
            for item in detected
        ]
        decision = _local_decision(detected, destination)

        masked_text: Optional[str] = None
        if decision == Decision.ALLOWED_WITH_MASKING:
            masked_text = _mask_local(text)

        summary = _summarize(decision, findings)
        return PreflightResult(
            decision=decision,
            summary=summary,
            masked_text=masked_text,
            findings=findings,
            mode_used="local",
            request_id=f"local_{uuid.uuid4().hex[:12]}",
        )

    def _run_cloud(
        self, *, text: str, destination: str, actor_id: Optional[str] = None
    ) -> PreflightResult:
        """Call the versioned server endpoint and deserialize the response.

        Sends ``source_type="sdk"`` so audit rows attribute the traffic
        correctly (distinguishes SDK calls from raw REST clients). Sends
        ``actor_id`` when the caller provided one.
        """
        body: dict = {"text": text, "destination": destination, "source_type": "sdk"}
        if actor_id:
            body["actor_id"] = actor_id
        try:
            payload, request_id = self._transport.request(
                method="POST",
                path=_V1_PREFLIGHT_PATH,
                json=body,
            )
        except ZotniqError:
            raise

        # Server response shape aligns with PreflightResult, but historically
        # returned "detected" for findings and "masked_content" for masked_text.
        # Normalize both spellings for forward compatibility.
        findings = _findings_from_server(payload)
        masked_text = payload.get("masked_text") or payload.get("masked_content")

        return PreflightResult(
            decision=Decision(payload.get("decision", "ALLOWED")),
            summary=payload.get("summary", ""),
            masked_text=masked_text,
            findings=findings,
            mode_used="cloud",
            request_id=request_id or f"cloud_{uuid.uuid4().hex[:12]}",
            monitor_mode=bool(payload.get("monitor_mode", False)),
        )

check(text, destination, mode='auto', actor_id=None)

Run a preflight check and return the decision.

Parameters:

Name Type Description Default
text str

The candidate content. Never persisted server-side under any retention_mode (Bug F invariant); only decision metadata lands in preflight_audit.

 required
destination DestinationLike

Where the content is going. Drives which policy applies. Accepts Destination enum or a raw string.

 required
mode Mode

See module docstring for resolution.

 'auto'
actor_id Optional[str]

Who inside your app made this call. Populates the actor_id column on the server-side audit row so dashboard filters ("show me every decision for [email protected]") work end-to-end. Common values: end-user email, internal user id, service account label. Local-mode calls carry it on the response for parity but never leave the process.

 None

Returns:

Type Description
PreflightResult

A PreflightResult with decision, findings, masked_text
PreflightResult

(when decision is ALLOWED_WITH_MASKING), and a request_id
PreflightResult

for correlation with server-side audit rows.

Raises:

Type Description
AuthError

mode="cloud" without an api_key configured.
NetworkError

cloud call failed and no fallback was requested.
ValidationError

server rejected the request body (rare).
Example

from zotniq import Zotniq client = Zotniq() result = client.preflight.check( ... "email me at [email protected]", ... destination="AI_TOOL", ... mode="local", ... ) result.decision.value 'ALLOWED_WITH_MASKING'

Source code in zotniq/resources/preflight.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 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
def check(
    self,
    text: str,
    destination: DestinationLike,
    mode: Mode = "auto",
    actor_id: Optional[str] = None,
) -> PreflightResult:
    """Run a preflight check and return the decision.

    Args:
        text: The candidate content. Never persisted server-side under
            any retention_mode (Bug F invariant); only decision metadata
            lands in ``preflight_audit``.
        destination: Where the content is going. Drives which policy
            applies. Accepts ``Destination`` enum or a raw string.
        mode: See module docstring for resolution.
        actor_id: Who inside your app made this call. Populates the
            ``actor_id`` column on the server-side audit row so
            dashboard filters ("show me every decision for
            [email protected]") work end-to-end. Common values: end-user
            email, internal user id, service account label. Local-mode
            calls carry it on the response for parity but never leave
            the process.

    Returns:
        A ``PreflightResult`` with decision, findings, masked_text
        (when decision is ``ALLOWED_WITH_MASKING``), and a request_id
        for correlation with server-side audit rows.

    Raises:
        AuthError: mode="cloud" without an api_key configured.
        NetworkError: cloud call failed and no fallback was requested.
        ValidationError: server rejected the request body (rare).

    Example:
        >>> from zotniq import Zotniq
        >>> client = Zotniq()
        >>> result = client.preflight.check(
        ...     "email me at [email protected]",
        ...     destination="AI_TOOL",
        ...     mode="local",
        ... )
        >>> result.decision.value
        'ALLOWED_WITH_MASKING'
    """
    dest_str = destination.value if isinstance(destination, Destination) else str(destination)
    resolved_mode = self._resolve_mode(mode)

    if resolved_mode == "local":
        result = self._run_local(text=text, destination=dest_str)
    elif resolved_mode == "cloud":
        result = self._run_cloud(text=text, destination=dest_str, actor_id=actor_id)
    else:
        # cloud_with_fallback — attempt cloud, fall through to local on NetworkError
        try:
            result = self._run_cloud(text=text, destination=dest_str, actor_id=actor_id)
        except NetworkError as e:
            logger.warning("cloud call failed, falling back to local: %s", e)
            result = self._run_local(text=text, destination=dest_str)

    self._fire_hook(text=text, destination=dest_str, result=result)
    return result

zotniq.resources.preflight.AsyncPreflightResource

Async twin of PreflightResource. Same public API, awaitable calls.

Source code in zotniq/resources/preflight.py 
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
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
class AsyncPreflightResource:
    """Async twin of PreflightResource. Same public API, awaitable calls."""

    def __init__(
        self,
        transport: AsyncHTTPTransport,
        api_key: Optional[str],
        hooks: Optional[HookDispatcher] = None,
        api_key_fingerprint: str = "(no key)",
    ) -> None:
        self._transport = transport
        self._api_key = api_key
        self._hooks = hooks
        self._api_key_fingerprint = api_key_fingerprint

    async def check(
        self,
        text: str,
        destination: DestinationLike,
        mode: Mode = "auto",
        actor_id: Optional[str] = None,
    ) -> PreflightResult:
        """Async twin of PreflightResource.check. Same semantics + actor_id."""
        dest_str = destination.value if isinstance(destination, Destination) else str(destination)
        resolved_mode = _resolve_mode_shared(mode, self._api_key)

        if resolved_mode == "local":
            result = _run_local_shared(text=text, destination=dest_str)
        elif resolved_mode == "cloud":
            result = await self._run_cloud(text=text, destination=dest_str, actor_id=actor_id)
        else:
            try:
                result = await self._run_cloud(text=text, destination=dest_str, actor_id=actor_id)
            except NetworkError as e:
                logger.warning("cloud call failed, falling back to local: %s", e)
                result = _run_local_shared(text=text, destination=dest_str)

        if self._hooks is not None:
            ctx = build_context(
                text=text,
                destination=dest_str,
                result=result,
                api_key_fingerprint=self._api_key_fingerprint,
            )
            self._hooks.emit(_HookEvent(result=result, context=ctx))
        return result

    async def _run_cloud(
        self, *, text: str, destination: str, actor_id: Optional[str] = None
    ) -> PreflightResult:
        body: dict = {"text": text, "destination": destination, "source_type": "sdk"}
        if actor_id:
            body["actor_id"] = actor_id
        payload, request_id = await self._transport.request(
            method="POST",
            path=_V1_PREFLIGHT_PATH,
            json=body,
        )
        findings = _findings_from_server(payload)
        masked_text = payload.get("masked_text") or payload.get("masked_content")
        return PreflightResult(
            decision=Decision(payload.get("decision", "ALLOWED")),
            summary=payload.get("summary", ""),
            masked_text=masked_text,
            findings=findings,
            mode_used="cloud",
            request_id=request_id or f"cloud_{uuid.uuid4().hex[:12]}",
            monitor_mode=bool(payload.get("monitor_mode", False)),
        )

check(text, destination, mode='auto', actor_id=None) async

Async twin of PreflightResource.check. Same semantics + actor_id.

Source code in zotniq/resources/preflight.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
async def check(
    self,
    text: str,
    destination: DestinationLike,
    mode: Mode = "auto",
    actor_id: Optional[str] = None,
) -> PreflightResult:
    """Async twin of PreflightResource.check. Same semantics + actor_id."""
    dest_str = destination.value if isinstance(destination, Destination) else str(destination)
    resolved_mode = _resolve_mode_shared(mode, self._api_key)

    if resolved_mode == "local":
        result = _run_local_shared(text=text, destination=dest_str)
    elif resolved_mode == "cloud":
        result = await self._run_cloud(text=text, destination=dest_str, actor_id=actor_id)
    else:
        try:
            result = await self._run_cloud(text=text, destination=dest_str, actor_id=actor_id)
        except NetworkError as e:
            logger.warning("cloud call failed, falling back to local: %s", e)
            result = _run_local_shared(text=text, destination=dest_str)

    if self._hooks is not None:
        ctx = build_context(
            text=text,
            destination=dest_str,
            result=result,
            api_key_fingerprint=self._api_key_fingerprint,
        )
        self._hooks.emit(_HookEvent(result=result, context=ctx))
    return result

Types

zotniq.types.Decision

Bases: str, Enum

Policy decision returned by client.preflight.check().

  • ALLOWED: no sensitive data detected, safe to send.
  • ALLOWED_WITH_MASKING: sensitive data found and masked; send result.masked_text instead of the original.
  • BLOCKED: policy forbids sending this content to this destination.
Source code in zotniq/types.py 
22
23
24
25
26
27
28
29
30
31
32
33
class Decision(str, Enum):
    """Policy decision returned by ``client.preflight.check()``.

    - ``ALLOWED``: no sensitive data detected, safe to send.
    - ``ALLOWED_WITH_MASKING``: sensitive data found and masked; send
      ``result.masked_text`` instead of the original.
    - ``BLOCKED``: policy forbids sending this content to this destination.
    """

    ALLOWED = "ALLOWED"
    ALLOWED_WITH_MASKING = "ALLOWED_WITH_MASKING"
    BLOCKED = "BLOCKED"

zotniq.types.Destination

Bases: str, Enum

Where the content is headed. Drives which policy applies.

Source code in zotniq/types.py 
36
37
38
39
40
41
class Destination(str, Enum):
    """Where the content is headed. Drives which policy applies."""

    AI_TOOL = "AI_TOOL"
    VENDOR = "VENDOR"
    CUSTOMER = "CUSTOMER"

zotniq.types.Finding

Bases: BaseModel

One aggregated detection finding attached to a PreflightResult.

Superset of the four write-path shapes (post-Bug H server unification): type is always populated; location/action/confidence are present when the underlying detector supplied them.

Source code in zotniq/types.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class Finding(BaseModel):
    """One aggregated detection finding attached to a PreflightResult.

    Superset of the four write-path shapes (post-Bug H server unification):
    ``type`` is always populated; ``location``/``action``/``confidence`` are
    present when the underlying detector supplied them.
    """

    model_config = ConfigDict(frozen=True)

    type: str
    count: int
    sample: Optional[str] = None
    location: Optional[str] = None
    action: Optional[str] = None
    confidence: Optional[float] = None

zotniq.types.PreflightResult

Bases: BaseModel

Full decision + findings returned by client.preflight.check().

masked_text is populated only when decision == ALLOWED_WITH_MASKING. request_id is populated for cloud calls (echoed from the server's x-request-id header) and set to a local UUID for local-mode calls, so every decision is correlatable.

monitor_mode reflects observe-only enforcement. When True, decision will be ALLOWED (never blocked or masked in the wire response) but the server-side audit row captured the true decision that would have applied under enforcement. Check this flag before assuming ALLOWED means the payload was actually clean.

Source code in zotniq/types.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class PreflightResult(BaseModel):
    """Full decision + findings returned by ``client.preflight.check()``.

    ``masked_text`` is populated only when ``decision == ALLOWED_WITH_MASKING``.
    ``request_id`` is populated for cloud calls (echoed from the server's
    ``x-request-id`` header) and set to a local UUID for local-mode calls,
    so every decision is correlatable.

    ``monitor_mode`` reflects observe-only enforcement. When True, ``decision``
    will be ``ALLOWED`` (never blocked or masked in the wire response) but the
    server-side audit row captured the *true* decision that would have applied
    under enforcement. Check this flag before assuming ``ALLOWED`` means the
    payload was actually clean.
    """

    model_config = ConfigDict(frozen=True)

    decision: Decision
    summary: str
    masked_text: Optional[str] = None
    findings: list[Finding] = Field(default_factory=list)
    mode_used: Literal["local", "cloud"] = "local"
    request_id: Optional[str] = None
    monitor_mode: bool = False

Errors

zotniq.errors.ZotniqError

Bases: Exception

Base class for all Zotniq SDK errors.

Source code in zotniq/errors.py 
12
13
class ZotniqError(Exception):
    """Base class for all Zotniq SDK errors."""

zotniq.errors.AuthError

Bases: ZotniqError

Missing / invalid API key, or 401/403 from the server.

Source code in zotniq/errors.py 
16
17
class AuthError(ZotniqError):
    """Missing / invalid API key, or 401/403 from the server."""

zotniq.errors.RateLimitError

Bases: ZotniqError

429 from the server. Retry-After honored automatically up to max_retries.

Source code in zotniq/errors.py 
20
21
22
23
24
25
class RateLimitError(ZotniqError):
    """429 from the server. Retry-After honored automatically up to max_retries."""

    def __init__(self, message: str, retry_after: Optional[float] = None) -> None:
        super().__init__(message)
        self.retry_after = retry_after

zotniq.errors.ValidationError

Bases: ZotniqError

4xx malformed request. Usually a client-side bug in the calling code.

Source code in zotniq/errors.py 
28
29
class ValidationError(ZotniqError):
    """4xx malformed request. Usually a client-side bug in the calling code."""

zotniq.errors.NetworkError

Bases: ZotniqError

5xx / timeout / DNS failure during a cloud call.

Never falls back to local silently — the caller sees the failure and decides. Set mode="cloud_with_fallback" per-call to opt into automatic local fallback instead.

Source code in zotniq/errors.py 
32
33
34
35
36
37
38
class NetworkError(ZotniqError):
    """5xx / timeout / DNS failure during a cloud call.

    Never falls back to local silently — the caller sees the failure and
    decides. Set ``mode="cloud_with_fallback"`` per-call to opt into
    automatic local fallback instead.
    """

Detection primitives (zero-dep)

zotniq.detection.detect(content, include_samples=True)

Main detection entry point.

Source code in zotniq/detection/detector.py 
337
338
339
340
def detect(content: str, include_samples: bool = True) -> list[DetectedItem]:
    """Main detection entry point."""
    findings = detect_patterns(content)
    return aggregate_findings(findings, include_samples=include_samples)

zotniq.detection.mask_text(text, findings=None)

Mask all sensitive data in text.

Detects and masks all sensitive data in the input text, replacing each occurrence with its masked equivalent.

Parameters:

Name Type Description Default
text str

The text to mask

 required
findings Optional[list[Finding]]

Optional pre-computed findings (if None, will detect)

 None

Returns:

Type Description
str

Text with all sensitive data masked
Source code in zotniq/detection/masker.py 
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
def mask_text(text: str, findings: Optional[list[Finding]] = None) -> str:
    """Mask all sensitive data in text.

    Detects and masks all sensitive data in the input text,
    replacing each occurrence with its masked equivalent.

    Args:
        text: The text to mask
        findings: Optional pre-computed findings (if None, will detect)

    Returns:
        Text with all sensitive data masked
    """
    if findings is None:
        findings = detect_patterns(text)

    if not findings:
        return text

    # Sort findings by position (reverse order for replacement)
    sorted_findings = sorted(findings, key=lambda f: f.start, reverse=True)

    result = text
    for finding in sorted_findings:
        masked = mask_value(finding.value, finding.detection_type)
        result = result[: finding.start] + masked + result[finding.end :]

    return result

SIEM forwarders

zotniq.siem.WebhookForwarder

Bases: BaseForwarder

POST each event as JSON to url.

Parameters:

Name Type Description Default
url str

Full HTTPS endpoint. HTTP allowed but discouraged.

 required
headers Optional[dict]

Extra headers merged on every request. Use for bearer tokens, API keys, or SIEM-specific auth headers.

 None
timeout float

Per-request timeout in seconds. Default 10.

 10.0
Example

forwarder = WebhookForwarder( url="https://soc.acme.com/hooks/dlp", headers={"Authorization": "Bearer secret-token"}, ) client = Zotniq(api_key="zot_sk_...", on_decision=forwarder)

Source code in zotniq/siem/webhook.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class WebhookForwarder(BaseForwarder):
    """POST each event as JSON to ``url``.

    Args:
        url: Full HTTPS endpoint. HTTP allowed but discouraged.
        headers: Extra headers merged on every request. Use for bearer
            tokens, API keys, or SIEM-specific auth headers.
        timeout: Per-request timeout in seconds. Default 10.

    Example:
        forwarder = WebhookForwarder(
            url="https://soc.acme.com/hooks/dlp",
            headers={"Authorization": "Bearer secret-token"},
        )
        client = Zotniq(api_key="zot_sk_...", on_decision=forwarder)
    """

    def __init__(
        self,
        url: str,
        headers: Optional[dict] = None,
        timeout: float = 10.0,
    ) -> None:
        self.url = url
        self.headers = {"Content-Type": "application/json", **(headers or {})}
        self.timeout = timeout

    def _send(self, payload: bytes) -> None:
        with httpx.Client(timeout=self.timeout) as client:
            response = client.post(self.url, content=payload, headers=self.headers)
            response.raise_for_status()

zotniq.siem.SplunkForwarder

Bases: BaseForwarder

POST to Splunk HEC.

Parameters:

Name Type Description Default
url str

Full HEC endpoint including path, e.g. https://splunk.acme.com:8088/services/collector.

 required
token str

HEC token from Splunk's Data Inputs → HTTP Event Collector.

 required
source str

source field on every event. Default zotniq-sdk.

 'zotniq-sdk'
sourcetype str

sourcetype field. Default zotniq:decision.

 'zotniq:decision'
index Optional[str]

Optional Splunk index override.

 None
timeout float

Per-request timeout. Default 10s.

 10.0
verify_ssl bool

Set False for self-signed HEC. Default True.

 True
Example

forwarder = SplunkForwarder( url="https://splunk.acme.com:8088/services/collector", token="hec-token-here", index="dlp_events", ) client = Zotniq(api_key="zot_sk_...", on_decision=forwarder)

Source code in zotniq/siem/splunk.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class SplunkForwarder(BaseForwarder):
    """POST to Splunk HEC.

    Args:
        url: Full HEC endpoint including path, e.g.
            ``https://splunk.acme.com:8088/services/collector``.
        token: HEC token from Splunk's Data Inputs → HTTP Event Collector.
        source: ``source`` field on every event. Default ``zotniq-sdk``.
        sourcetype: ``sourcetype`` field. Default ``zotniq:decision``.
        index: Optional Splunk index override.
        timeout: Per-request timeout. Default 10s.
        verify_ssl: Set False for self-signed HEC. Default True.

    Example:
        forwarder = SplunkForwarder(
            url="https://splunk.acme.com:8088/services/collector",
            token="hec-token-here",
            index="dlp_events",
        )
        client = Zotniq(api_key="zot_sk_...", on_decision=forwarder)
    """

    def __init__(
        self,
        url: str,
        token: str,
        source: str = "zotniq-sdk",
        sourcetype: str = "zotniq:decision",
        index: Optional[str] = None,
        timeout: float = 10.0,
        verify_ssl: bool = True,
    ) -> None:
        self.url = url
        self.token = token
        self.source = source
        self.sourcetype = sourcetype
        self.index = index
        self.timeout = timeout
        self.verify_ssl = verify_ssl

    def __call__(self, result: PreflightResult, context: dict) -> None:
        try:
            event = self._build_event(result, context)
            envelope: dict[str, Any] = {
                "event": event,
                "source": self.source,
                "sourcetype": self.sourcetype,
            }
            if self.index:
                envelope["index"] = self.index
            payload = json.dumps(envelope).encode("utf-8")
            self._send(payload)
        except Exception:
            # Base class semantics — never raise to caller.
            import logging

            logging.getLogger(__name__).warning("SplunkForwarder failed", exc_info=True)

    def _send(self, payload: bytes) -> None:
        headers = {
            "Authorization": f"Splunk {self.token}",
            "Content-Type": "application/json",
        }
        with httpx.Client(timeout=self.timeout, verify=self.verify_ssl) as client:
            response = client.post(self.url, content=payload, headers=headers)
            response.raise_for_status()

zotniq.siem.DatadogForwarder

Bases: BaseForwarder

POST to Datadog Logs API.

Parameters:

Name Type Description Default
api_key str

Datadog API key (not app key).

 required
site str

Datadog region. Default datadoghq.com.

 'datadoghq.com'
ddsource str

Datadog ddsource field. Default zotniq.

 'zotniq'
service str

Datadog service field. Default dlp.

 'dlp'
tags Optional[list[str]]

Extra tags appended to every event.

 None
timeout float

Per-request timeout. Default 10s.

 10.0
Example

forwarder = DatadogForwarder( api_key="dd-api-key", site="datadoghq.com", service="chat-backend", tags=["env:prod", "team:security"], ) client = Zotniq(api_key="zot_sk_...", on_decision=forwarder)

Source code in zotniq/siem/datadog.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class DatadogForwarder(BaseForwarder):
    """POST to Datadog Logs API.

    Args:
        api_key: Datadog API key (not app key).
        site: Datadog region. Default ``datadoghq.com``.
        ddsource: Datadog ``ddsource`` field. Default ``zotniq``.
        service: Datadog ``service`` field. Default ``dlp``.
        tags: Extra tags appended to every event.
        timeout: Per-request timeout. Default 10s.

    Example:
        forwarder = DatadogForwarder(
            api_key="dd-api-key",
            site="datadoghq.com",
            service="chat-backend",
            tags=["env:prod", "team:security"],
        )
        client = Zotniq(api_key="zot_sk_...", on_decision=forwarder)
    """

    def __init__(
        self,
        api_key: str,
        site: str = "datadoghq.com",
        ddsource: str = "zotniq",
        service: str = "dlp",
        tags: Optional[list[str]] = None,
        timeout: float = 10.0,
    ) -> None:
        self.api_key = api_key
        self.site = site
        self.ddsource = ddsource
        self.service = service
        self.tags = tags or []
        self.timeout = timeout
        self._intake_url = _SITE_TO_INTAKE.get(site, f"https://http-intake.logs.{site}")

    def __call__(self, result: PreflightResult, context: dict) -> None:
        try:
            event = self._build_event(result, context)
            envelope: dict[str, Any] = {
                "ddsource": self.ddsource,
                "service": self.service,
                "message": event,
            }
            if self.tags:
                envelope["ddtags"] = ",".join(self.tags)
            payload = json.dumps([envelope]).encode("utf-8")  # Datadog expects a list
            self._send(payload)
        except Exception:
            import logging

            logging.getLogger(__name__).warning("DatadogForwarder failed", exc_info=True)

    def _send(self, payload: bytes) -> None:
        headers = {
            "DD-API-KEY": self.api_key,
            "Content-Type": "application/json",
        }
        with httpx.Client(timeout=self.timeout) as client:
            response = client.post(
                f"{self._intake_url}/api/v2/logs", content=payload, headers=headers
            )
            response.raise_for_status()

zotniq.siem.FileForwarder

Bases: BaseForwarder

Append events as newline-delimited JSON to a local file.

Parameters:

Name Type Description Default
path Union[str, PathLike]

Destination file path. Parent directory must exist.

 required
format str

Only "ndjson" supported today.

 'ndjson'
Example

forwarder = FileForwarder(path="/var/log/zotniq/decisions.ndjson") client = Zotniq(api_key="zot_sk_...", on_decision=forwarder)

Concurrency: writes are serialized via a file-level lock so multiple Zotniq clients in the same process can share a target file without interleaving lines.

Source code in zotniq/siem/file.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class FileForwarder(BaseForwarder):
    """Append events as newline-delimited JSON to a local file.

    Args:
        path: Destination file path. Parent directory must exist.
        format: Only ``"ndjson"`` supported today.

    Example:
        forwarder = FileForwarder(path="/var/log/zotniq/decisions.ndjson")
        client = Zotniq(api_key="zot_sk_...", on_decision=forwarder)

    Concurrency: writes are serialized via a file-level lock so multiple
    Zotniq clients in the same process can share a target file without
    interleaving lines.
    """

    _lock = threading.Lock()

    def __init__(self, path: Union[str, os.PathLike], format: str = "ndjson") -> None:
        if format != "ndjson":
            raise ValueError(f"unsupported format={format!r}; only 'ndjson' supported in v0.1")
        self.path = Path(path)
        if not self.path.parent.exists():
            raise FileNotFoundError(
                f"parent directory {self.path.parent} does not exist. "
                "Create it before instantiating FileForwarder."
            )

    def _send(self, payload: bytes) -> None:
        with self._lock, open(self.path, "ab") as f:
            f.write(payload)
            f.write(b"\n")

OpenAI integration

zotniq.integrations.openai.wrap_openai(zotniq_client, **openai_kwargs)

Return an OpenAI-compatible client that runs preflight on user messages.

Parameters:

Name Type Description Default
zotniq_client Zotniq

A Zotniq client. Determines mode, api_key, SIEM.

 required
**openai_kwargs Any

Passed through to the underlying openai.OpenAI constructor. Typically api_key="sk-...".

 {}

Raises:

Type Description
ImportError

openai package not installed. Install with pip install zotniq[openai].
Source code in zotniq/integrations/openai.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def wrap_openai(zotniq_client: Zotniq, **openai_kwargs: Any) -> Any:
    """Return an OpenAI-compatible client that runs preflight on user messages.

    Args:
        zotniq_client: A ``Zotniq`` client. Determines mode, api_key, SIEM.
        **openai_kwargs: Passed through to the underlying ``openai.OpenAI``
            constructor. Typically ``api_key="sk-..."``.

    Raises:
        ImportError: openai package not installed. Install with
            ``pip install zotniq[openai]``.
    """
    try:
        import openai as _openai
    except ImportError as e:  # pragma: no cover
        raise ImportError(
            "The openai package is required for zotniq.integrations.openai. "
            "Install with: pip install zotniq[openai]"
        ) from e

    inner = _openai.OpenAI(**openai_kwargs)
    return _WrappedOpenAI(inner=inner, zotniq=zotniq_client)