Skip to content

Core Types

Core data models, enumerations, configuration classes, and exceptions for rotalabs-comply.

Enumerations

RiskLevel

RiskLevel

Risk severity levels for compliance classification.

These levels are used to categorize the severity of compliance violations and to set risk thresholds for AI systems.

Attributes:

Name Type Description
LOW

Minor risk with minimal compliance impact.
MEDIUM

Moderate risk requiring attention.
HIGH

Significant risk requiring immediate action.
CRITICAL

Severe risk with potential regulatory consequences.
Source code in src/rotalabs_comply/core/types.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class RiskLevel(str, Enum):
    """
    Risk severity levels for compliance classification.

    These levels are used to categorize the severity of compliance
    violations and to set risk thresholds for AI systems.

    Attributes:
        LOW: Minor risk with minimal compliance impact.
        MEDIUM: Moderate risk requiring attention.
        HIGH: Significant risk requiring immediate action.
        CRITICAL: Severe risk with potential regulatory consequences.
    """

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

LOW class-attribute instance-attribute 

LOW = 'low'

MEDIUM class-attribute instance-attribute 

MEDIUM = 'medium'

HIGH class-attribute instance-attribute 

HIGH = 'high'

CRITICAL class-attribute instance-attribute 

CRITICAL = 'critical'

Risk severity levels for compliance classification.

Level Value Description
LOW "low" Minor risk with minimal compliance impact
MEDIUM "medium" Moderate risk requiring attention
HIGH "high" Significant risk requiring immediate action
CRITICAL "critical" Severe risk with potential regulatory consequences

Example:

from rotalabs_comply import RiskLevel

level = RiskLevel.HIGH
print(level.value)  # "high"

Framework

Framework

Supported regulatory compliance frameworks.

Each framework represents a set of compliance requirements that can be validated against AI system operations.

Attributes:

Name Type Description
EU_AI_ACT

European Union AI Act requirements.
SOC2

Service Organization Control 2 security standards.
HIPAA

Health Insurance Portability and Accountability Act.
GDPR

General Data Protection Regulation.
NIST_AI_RMF

NIST AI Risk Management Framework.
ISO_42001

ISO/IEC 42001 AI Management System standard.
MAS

Monetary Authority of Singapore FEAT principles.
Source code in src/rotalabs_comply/core/types.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class Framework(str, Enum):
    """
    Supported regulatory compliance frameworks.

    Each framework represents a set of compliance requirements
    that can be validated against AI system operations.

    Attributes:
        EU_AI_ACT: European Union AI Act requirements.
        SOC2: Service Organization Control 2 security standards.
        HIPAA: Health Insurance Portability and Accountability Act.
        GDPR: General Data Protection Regulation.
        NIST_AI_RMF: NIST AI Risk Management Framework.
        ISO_42001: ISO/IEC 42001 AI Management System standard.
        MAS: Monetary Authority of Singapore FEAT principles.
    """

    EU_AI_ACT = "eu_ai_act"
    SOC2 = "soc2"
    HIPAA = "hipaa"
    GDPR = "gdpr"
    NIST_AI_RMF = "nist_ai_rmf"
    ISO_42001 = "iso_42001"
    MAS = "mas"

Supported regulatory compliance frameworks.

Framework Value Description
EU_AI_ACT "eu_ai_act" European Union AI Act requirements
SOC2 "soc2" Service Organization Control 2 standards
HIPAA "hipaa" Health Insurance Portability and Accountability Act
GDPR "gdpr" General Data Protection Regulation
NIST_AI_RMF "nist_ai_rmf" NIST AI Risk Management Framework
ISO_42001 "iso_42001" ISO/IEC 42001 AI Management System

Example:

from rotalabs_comply import Framework

frameworks = [Framework.EU_AI_ACT, Framework.HIPAA]

Data Models

AuditEntry

AuditEntry

A single audit log entry capturing an AI system interaction.

This model captures comprehensive information about AI model invocations, including input/output data (or hashes for privacy), safety checks, and performance metrics.

Attributes:

Name Type Description
id str

Unique identifier for this audit entry (UUID).
timestamp datetime

When the interaction occurred.
provider Optional[str]

AI provider name (e.g., "openai", "anthropic").
model Optional[str]

Model identifier (e.g., "gpt-4", "claude-3-opus").
conversation_id Optional[str]

Optional ID linking related interactions.
input_hash str

SHA-256 hash of the input content.
output_hash str

SHA-256 hash of the output content.
input_content Optional[str]

Actual input text (only if store_content enabled).
output_content Optional[str]

Actual output text (only if store_content enabled).
safety_passed bool

Whether all safety checks passed.
detectors_triggered List[str]

List of detector names that flagged content.
block_reason Optional[str]

Reason if the interaction was blocked.
alerts List[str]

List of alert messages generated.
latency_ms float

Response time in milliseconds.
input_tokens Optional[int]

Number of input tokens (if available).
output_tokens Optional[int]

Number of output tokens (if available).
metadata Dict[str, Any]

Additional custom metadata.
Example

entry = AuditEntry( ... provider="openai", ... model="gpt-4", ... input_hash="abc123...", ... output_hash="def456...", ... safety_passed=True, ... detectors_triggered=[], ... latency_ms=245.5, ... )

Source code in src/rotalabs_comply/core/types.py 
 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
class AuditEntry(BaseModel):
    """
    A single audit log entry capturing an AI system interaction.

    This model captures comprehensive information about AI model invocations,
    including input/output data (or hashes for privacy), safety checks,
    and performance metrics.

    Attributes:
        id: Unique identifier for this audit entry (UUID).
        timestamp: When the interaction occurred.
        provider: AI provider name (e.g., "openai", "anthropic").
        model: Model identifier (e.g., "gpt-4", "claude-3-opus").
        conversation_id: Optional ID linking related interactions.
        input_hash: SHA-256 hash of the input content.
        output_hash: SHA-256 hash of the output content.
        input_content: Actual input text (only if store_content enabled).
        output_content: Actual output text (only if store_content enabled).
        safety_passed: Whether all safety checks passed.
        detectors_triggered: List of detector names that flagged content.
        block_reason: Reason if the interaction was blocked.
        alerts: List of alert messages generated.
        latency_ms: Response time in milliseconds.
        input_tokens: Number of input tokens (if available).
        output_tokens: Number of output tokens (if available).
        metadata: Additional custom metadata.

    Example:
        >>> entry = AuditEntry(
        ...     provider="openai",
        ...     model="gpt-4",
        ...     input_hash="abc123...",
        ...     output_hash="def456...",
        ...     safety_passed=True,
        ...     detectors_triggered=[],
        ...     latency_ms=245.5,
        ... )
    """

    model_config = ConfigDict(
        populate_by_name=True,
        use_enum_values=True,
        validate_default=True,
    )

    id: str = Field(
        default_factory=lambda: str(uuid4()),
        description="Unique identifier for this audit entry",
    )
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        description="When the interaction occurred",
    )
    provider: Optional[str] = Field(
        default=None,
        description="AI provider name (e.g., 'openai', 'anthropic')",
    )
    model: Optional[str] = Field(
        default=None,
        description="Model identifier (e.g., 'gpt-4', 'claude-3-opus')",
    )
    conversation_id: Optional[str] = Field(
        default=None,
        description="Optional ID linking related interactions",
    )
    input_hash: str = Field(
        ...,
        description="SHA-256 hash of the input content",
    )
    output_hash: str = Field(
        ...,
        description="SHA-256 hash of the output content",
    )
    input_content: Optional[str] = Field(
        default=None,
        description="Actual input text (only if store_content enabled)",
    )
    output_content: Optional[str] = Field(
        default=None,
        description="Actual output text (only if store_content enabled)",
    )
    safety_passed: bool = Field(
        ...,
        description="Whether all safety checks passed",
    )
    detectors_triggered: List[str] = Field(
        default_factory=list,
        description="List of detector names that flagged content",
    )
    block_reason: Optional[str] = Field(
        default=None,
        description="Reason if the interaction was blocked",
    )
    alerts: List[str] = Field(
        default_factory=list,
        description="List of alert messages generated",
    )
    latency_ms: float = Field(
        ...,
        ge=0,
        description="Response time in milliseconds",
    )
    input_tokens: Optional[int] = Field(
        default=None,
        ge=0,
        description="Number of input tokens (if available)",
    )
    output_tokens: Optional[int] = Field(
        default=None,
        ge=0,
        description="Number of output tokens (if available)",
    )
    metadata: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional custom metadata",
    )

A single audit log entry capturing an AI system interaction.

Attributes:

Attribute Type Description
id str Unique identifier (UUID, auto-generated)
timestamp datetime When the interaction occurred
provider Optional[str] AI provider name (e.g., "openai")
model Optional[str] Model identifier (e.g., "gpt-4")
conversation_id Optional[str] ID linking related interactions
input_hash str SHA-256 hash of input content
output_hash str SHA-256 hash of output content
input_content Optional[str] Actual input (if stored, may be encrypted)
output_content Optional[str] Actual output (if stored, may be encrypted)
safety_passed bool Whether all safety checks passed
detectors_triggered List[str] Safety detectors that flagged content
block_reason Optional[str] Reason if interaction was blocked
alerts List[str] Alert messages generated
latency_ms float Response time in milliseconds
input_tokens Optional[int] Number of input tokens
output_tokens Optional[int] Number of output tokens
metadata Dict[str, Any] Additional custom metadata

Example:

from rotalabs_comply import AuditEntry

entry = AuditEntry(
    provider="openai",
    model="gpt-4",
    input_hash="abc123...",
    output_hash="def456...",
    safety_passed=True,
    latency_ms=245.5,
)

ComplianceProfile

ComplianceProfile

Configuration profile defining compliance requirements.

This model specifies which regulatory frameworks apply, risk tolerance, documentation requirements, and data handling policies for an AI system.

Attributes:

Name Type Description
frameworks List[Framework]

List of regulatory frameworks to comply with.
risk_level RiskLevel

Maximum acceptable risk level.
required_documentation bool

Whether comprehensive docs are required.
data_retention_days int

How long to retain audit data.
encrypt_audit_logs bool

Whether to encrypt stored audit logs.
store_content bool

Whether to store actual content vs just hashes.
custom_policies Dict[str, Any]

Additional custom policy configurations.
Example

profile = ComplianceProfile( ... frameworks=[Framework.GDPR, Framework.EU_AI_ACT], ... risk_level=RiskLevel.MEDIUM, ... data_retention_days=365, ... store_content=False, # Privacy mode ... )

Source code in src/rotalabs_comply/core/types.py 
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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
class ComplianceProfile(BaseModel):
    """
    Configuration profile defining compliance requirements.

    This model specifies which regulatory frameworks apply, risk tolerance,
    documentation requirements, and data handling policies for an AI system.

    Attributes:
        frameworks: List of regulatory frameworks to comply with.
        risk_level: Maximum acceptable risk level.
        required_documentation: Whether comprehensive docs are required.
        data_retention_days: How long to retain audit data.
        encrypt_audit_logs: Whether to encrypt stored audit logs.
        store_content: Whether to store actual content vs just hashes.
        custom_policies: Additional custom policy configurations.

    Example:
        >>> profile = ComplianceProfile(
        ...     frameworks=[Framework.GDPR, Framework.EU_AI_ACT],
        ...     risk_level=RiskLevel.MEDIUM,
        ...     data_retention_days=365,
        ...     store_content=False,  # Privacy mode
        ... )
    """

    model_config = ConfigDict(
        populate_by_name=True,
        use_enum_values=True,
        validate_default=True,
    )

    frameworks: List[Framework] = Field(
        default_factory=list,
        description="List of regulatory frameworks to comply with",
    )
    risk_level: RiskLevel = Field(
        default=RiskLevel.MEDIUM,
        description="Maximum acceptable risk level",
    )
    required_documentation: bool = Field(
        default=True,
        description="Whether comprehensive documentation is required",
    )
    data_retention_days: int = Field(
        default=365,
        ge=1,
        description="How long to retain audit data in days",
    )
    encrypt_audit_logs: bool = Field(
        default=True,
        description="Whether to encrypt stored audit logs",
    )
    store_content: bool = Field(
        default=False,
        description="Whether to store actual content vs just hashes",
    )
    custom_policies: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional custom policy configurations",
    )

Configuration profile defining compliance requirements.

Attributes:

Attribute Type Default Description
frameworks List[Framework] [] Regulatory frameworks to comply with
risk_level RiskLevel MEDIUM Maximum acceptable risk level
required_documentation bool True Whether comprehensive docs required
data_retention_days int 365 Days to retain audit data
encrypt_audit_logs bool True Whether to encrypt stored logs
store_content bool False Store content vs just hashes
custom_policies Dict[str, Any] {} Custom policy configurations

Example:

from rotalabs_comply import ComplianceProfile, Framework, RiskLevel

profile = ComplianceProfile(
    frameworks=[Framework.GDPR, Framework.EU_AI_ACT],
    risk_level=RiskLevel.MEDIUM,
    data_retention_days=365,
    store_content=False,
)

ComplianceViolation

ComplianceViolation

A single compliance violation detected during checking.

This model captures details about a specific compliance rule violation, including the framework, severity, and recommended remediation steps.

Attributes:

Name Type Description
framework Framework

The regulatory framework that was violated.
rule_id str

Identifier of the specific rule that was violated.
severity RiskLevel

How severe the violation is.
description str

Human-readable description of the violation.
evidence Dict[str, Any]

Data supporting the violation finding.
remediation str

Recommended steps to fix the violation.
timestamp datetime

When the violation was detected.
Example

violation = ComplianceViolation( ... framework=Framework.GDPR, ... rule_id="GDPR-ART13-1", ... severity=RiskLevel.HIGH, ... description="Personal data processed without consent record", ... evidence={"field": "user_email", "record_id": "123"}, ... remediation="Implement consent tracking mechanism", ... )

Source code in src/rotalabs_comply/core/types.py 
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
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
class ComplianceViolation(BaseModel):
    """
    A single compliance violation detected during checking.

    This model captures details about a specific compliance rule violation,
    including the framework, severity, and recommended remediation steps.

    Attributes:
        framework: The regulatory framework that was violated.
        rule_id: Identifier of the specific rule that was violated.
        severity: How severe the violation is.
        description: Human-readable description of the violation.
        evidence: Data supporting the violation finding.
        remediation: Recommended steps to fix the violation.
        timestamp: When the violation was detected.

    Example:
        >>> violation = ComplianceViolation(
        ...     framework=Framework.GDPR,
        ...     rule_id="GDPR-ART13-1",
        ...     severity=RiskLevel.HIGH,
        ...     description="Personal data processed without consent record",
        ...     evidence={"field": "user_email", "record_id": "123"},
        ...     remediation="Implement consent tracking mechanism",
        ... )
    """

    model_config = ConfigDict(
        populate_by_name=True,
        use_enum_values=True,
        validate_default=True,
    )

    framework: Framework = Field(
        ...,
        description="The regulatory framework that was violated",
    )
    rule_id: str = Field(
        ...,
        description="Identifier of the specific rule that was violated",
    )
    severity: RiskLevel = Field(
        ...,
        description="How severe the violation is",
    )
    description: str = Field(
        ...,
        description="Human-readable description of the violation",
    )
    evidence: Dict[str, Any] = Field(
        default_factory=dict,
        description="Data supporting the violation finding",
    )
    remediation: str = Field(
        ...,
        description="Recommended steps to fix the violation",
    )
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        description="When the violation was detected",
    )

A single compliance violation detected during checking.

Attributes:

Attribute Type Description
framework Framework The framework that was violated
rule_id str Identifier of the violated rule
severity RiskLevel Severity of the violation
description str Human-readable description
evidence Dict[str, Any] Data supporting the finding
remediation str Recommended fix steps
timestamp datetime When violation was detected

Example:

from rotalabs_comply import ComplianceViolation, Framework, RiskLevel

violation = ComplianceViolation(
    framework=Framework.GDPR,
    rule_id="GDPR-ART13-1",
    severity=RiskLevel.HIGH,
    description="Personal data processed without consent record",
    evidence={"field": "user_email"},
    remediation="Implement consent tracking mechanism",
)

ComplianceCheckResult

ComplianceCheckResult

Result of a compliance check against a regulatory framework.

This model captures the outcome of validating an AI system's operations against compliance requirements, including any violations found.

Attributes:

Name Type Description
passed bool

Whether the check passed (no critical violations).
framework Framework

The framework that was checked against.
violations List[ComplianceViolation]

List of compliance violations found.
warnings List[str]

Non-critical issues that should be addressed.
recommendations List[str]

Suggestions for improving compliance posture.
checked_at datetime

When the compliance check was performed.
Example

result = ComplianceCheckResult( ... passed=False, ... framework=Framework.SOC2, ... violations=[violation], ... warnings=["Audit log rotation not configured"], ... recommendations=["Enable encryption for audit logs"], ... )

Source code in src/rotalabs_comply/core/types.py 
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
class ComplianceCheckResult(BaseModel):
    """
    Result of a compliance check against a regulatory framework.

    This model captures the outcome of validating an AI system's operations
    against compliance requirements, including any violations found.

    Attributes:
        passed: Whether the check passed (no critical violations).
        framework: The framework that was checked against.
        violations: List of compliance violations found.
        warnings: Non-critical issues that should be addressed.
        recommendations: Suggestions for improving compliance posture.
        checked_at: When the compliance check was performed.

    Example:
        >>> result = ComplianceCheckResult(
        ...     passed=False,
        ...     framework=Framework.SOC2,
        ...     violations=[violation],
        ...     warnings=["Audit log rotation not configured"],
        ...     recommendations=["Enable encryption for audit logs"],
        ... )
    """

    model_config = ConfigDict(
        populate_by_name=True,
        use_enum_values=True,
        validate_default=True,
    )

    passed: bool = Field(
        ...,
        description="Whether the check passed (no critical violations)",
    )
    framework: Framework = Field(
        ...,
        description="The framework that was checked against",
    )
    violations: List[ComplianceViolation] = Field(
        default_factory=list,
        description="List of compliance violations found",
    )
    warnings: List[str] = Field(
        default_factory=list,
        description="Non-critical issues that should be addressed",
    )
    recommendations: List[str] = Field(
        default_factory=list,
        description="Suggestions for improving compliance posture",
    )
    checked_at: datetime = Field(
        default_factory=datetime.utcnow,
        description="When the compliance check was performed",
    )

Result of a compliance check against a regulatory framework.

Attributes:

Attribute Type Description
passed bool Whether check passed (no critical violations)
framework Framework Framework checked against
violations List[ComplianceViolation] Violations found
warnings List[str] Non-critical issues
recommendations List[str] Improvement suggestions
checked_at datetime When check was performed

Example:

from rotalabs_comply import ComplianceCheckResult, Framework

result = ComplianceCheckResult(
    passed=False,
    framework=Framework.SOC2,
    violations=[violation],
    warnings=["Audit log rotation not configured"],
    recommendations=["Enable encryption for audit logs"],
)

Configuration Classes

AuditConfig

AuditConfig

Configuration for audit logging behavior.

This model defines how audit entries are stored, encrypted, rotated, and retained over time.

Attributes:

Name Type Description
destination str

File path or S3 URL (s3://bucket/prefix) for audit logs.
encryption_enabled bool

Whether to encrypt audit log data at rest.
encryption_key Optional[str]

Base64-encoded encryption key (auto-generated if not provided).
retention_days int

Number of days to retain audit logs before deletion.
max_file_size_mb int

Maximum size of a single audit log file before rotation.
rotation_enabled bool

Whether to enable automatic log rotation.
compression_enabled bool

Whether to compress audit log files.
Example

config = AuditConfig( ... destination="/var/log/ai-audit/", ... encryption_enabled=True, ... retention_days=365, ... rotation_enabled=True, ... )

s3_config = AuditConfig( ... destination="s3://my-bucket/audit-logs/", ... encryption_enabled=True, ... compression_enabled=True, ... )

Source code in src/rotalabs_comply/core/config.py 
 14
 15
 16
 17
 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
 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
class AuditConfig(BaseModel):
    """
    Configuration for audit logging behavior.

    This model defines how audit entries are stored, encrypted,
    rotated, and retained over time.

    Attributes:
        destination: File path or S3 URL (s3://bucket/prefix) for audit logs.
        encryption_enabled: Whether to encrypt audit log data at rest.
        encryption_key: Base64-encoded encryption key (auto-generated if not provided).
        retention_days: Number of days to retain audit logs before deletion.
        max_file_size_mb: Maximum size of a single audit log file before rotation.
        rotation_enabled: Whether to enable automatic log rotation.
        compression_enabled: Whether to compress audit log files.

    Example:
        >>> config = AuditConfig(
        ...     destination="/var/log/ai-audit/",
        ...     encryption_enabled=True,
        ...     retention_days=365,
        ...     rotation_enabled=True,
        ... )

        >>> s3_config = AuditConfig(
        ...     destination="s3://my-bucket/audit-logs/",
        ...     encryption_enabled=True,
        ...     compression_enabled=True,
        ... )
    """

    model_config = ConfigDict(
        populate_by_name=True,
        validate_default=True,
        extra="forbid",
    )

    destination: str = Field(
        ...,
        description="File path or S3 URL (s3://bucket/prefix) for audit logs",
    )
    encryption_enabled: bool = Field(
        default=True,
        description="Whether to encrypt audit log data at rest",
    )
    encryption_key: Optional[str] = Field(
        default=None,
        description="Base64-encoded encryption key (auto-generated if not provided)",
    )
    retention_days: int = Field(
        default=365,
        ge=1,
        le=3650,  # Max 10 years
        description="Number of days to retain audit logs before deletion",
    )
    max_file_size_mb: int = Field(
        default=100,
        ge=1,
        le=1000,
        description="Maximum size of a single audit log file before rotation (MB)",
    )
    rotation_enabled: bool = Field(
        default=True,
        description="Whether to enable automatic log rotation",
    )
    compression_enabled: bool = Field(
        default=False,
        description="Whether to compress audit log files",
    )

    def model_post_init(self, __context) -> None:
        """Auto-generate encryption key if encryption is enabled but no key provided."""
        if self.encryption_enabled and self.encryption_key is None:
            # Generate a secure 256-bit key (32 bytes) as hex string
            object.__setattr__(self, "encryption_key", secrets.token_hex(32))

    @field_validator("destination")
    @classmethod
    def validate_destination(cls, v: str) -> str:
        """Validate that destination is a valid path or S3 URL."""
        v = v.strip()
        if not v:
            raise ValueError("destination cannot be empty")
        # Allow file paths (absolute or relative) and S3 URLs
        if v.startswith("s3://"):
            # Basic S3 URL validation
            parts = v[5:].split("/", 1)
            if not parts[0]:
                raise ValueError("S3 URL must include bucket name")
        return v

    @property
    def is_s3_destination(self) -> bool:
        """Check if the destination is an S3 URL."""
        return self.destination.startswith("s3://")

    @property
    def s3_bucket(self) -> Optional[str]:
        """Extract S3 bucket name from destination if applicable."""
        if not self.is_s3_destination:
            return None
        return self.destination[5:].split("/", 1)[0]

    @property
    def s3_prefix(self) -> Optional[str]:
        """Extract S3 key prefix from destination if applicable."""
        if not self.is_s3_destination:
            return None
        parts = self.destination[5:].split("/", 1)
        return parts[1] if len(parts) > 1 else ""

is_s3_destination property 

is_s3_destination: bool

Check if the destination is an S3 URL.

s3_bucket property 

s3_bucket: Optional[str]

Extract S3 bucket name from destination if applicable.

s3_prefix property 

s3_prefix: Optional[str]

Extract S3 key prefix from destination if applicable.

model_post_init 

model_post_init(__context) -> None

Auto-generate encryption key if encryption is enabled but no key provided.

Source code in src/rotalabs_comply/core/config.py 
84
85
86
87
88
def model_post_init(self, __context) -> None:
    """Auto-generate encryption key if encryption is enabled but no key provided."""
    if self.encryption_enabled and self.encryption_key is None:
        # Generate a secure 256-bit key (32 bytes) as hex string
        object.__setattr__(self, "encryption_key", secrets.token_hex(32))

validate_destination classmethod 

validate_destination(v: str) -> str

Validate that destination is a valid path or S3 URL.

Source code in src/rotalabs_comply/core/config.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@field_validator("destination")
@classmethod
def validate_destination(cls, v: str) -> str:
    """Validate that destination is a valid path or S3 URL."""
    v = v.strip()
    if not v:
        raise ValueError("destination cannot be empty")
    # Allow file paths (absolute or relative) and S3 URLs
    if v.startswith("s3://"):
        # Basic S3 URL validation
        parts = v[5:].split("/", 1)
        if not parts[0]:
            raise ValueError("S3 URL must include bucket name")
    return v

Configuration for audit logging behavior.

Attributes:

Attribute Type Default Description
destination str Required File path or S3 URL for logs
encryption_enabled bool True Encrypt data at rest
encryption_key Optional[str] Auto-gen Base64-encoded encryption key
retention_days int 365 Days before deletion (1-3650)
max_file_size_mb int 100 Max file size before rotation
rotation_enabled bool True Enable automatic rotation
compression_enabled bool False Compress audit files

Properties:

Property Type Description
is_s3_destination bool Whether destination is S3 URL
s3_bucket Optional[str] S3 bucket name if applicable
s3_prefix Optional[str] S3 key prefix if applicable

Example:

from rotalabs_comply import AuditConfig

# File storage
config = AuditConfig(
    destination="/var/log/ai-audit/",
    encryption_enabled=True,
    retention_days=365,
)

# S3 storage
s3_config = AuditConfig(
    destination="s3://my-bucket/audit-logs/",
    encryption_enabled=True,
    compression_enabled=True,
)

StorageConfig

StorageConfig

Configuration for storage backend selection and settings.

This model defines which storage backend to use and its specific configuration options.

Attributes:

Name Type Description
backend Literal['file', 's3', 'memory']

Storage backend type ("file", "s3", or "memory").
path Optional[str]

Local file path for file backend.
bucket Optional[str]

S3 bucket name for S3 backend.
prefix Optional[str]

S3 key prefix for organizing objects.
region Optional[str]

AWS region for S3 backend.
Example

file_storage = StorageConfig( ... backend="file", ... path="/var/log/ai-audit/", ... )

s3_storage = StorageConfig( ... backend="s3", ... bucket="my-audit-bucket", ... prefix="prod/audit-logs/", ... region="us-west-2", ... )

memory_storage = StorageConfig( ... backend="memory", # For testing ... )

Source code in src/rotalabs_comply/core/config.py 
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
class StorageConfig(BaseModel):
    """
    Configuration for storage backend selection and settings.

    This model defines which storage backend to use and its
    specific configuration options.

    Attributes:
        backend: Storage backend type ("file", "s3", or "memory").
        path: Local file path for file backend.
        bucket: S3 bucket name for S3 backend.
        prefix: S3 key prefix for organizing objects.
        region: AWS region for S3 backend.

    Example:
        >>> file_storage = StorageConfig(
        ...     backend="file",
        ...     path="/var/log/ai-audit/",
        ... )

        >>> s3_storage = StorageConfig(
        ...     backend="s3",
        ...     bucket="my-audit-bucket",
        ...     prefix="prod/audit-logs/",
        ...     region="us-west-2",
        ... )

        >>> memory_storage = StorageConfig(
        ...     backend="memory",  # For testing
        ... )
    """

    model_config = ConfigDict(
        populate_by_name=True,
        validate_default=True,
        extra="forbid",
    )

    backend: Literal["file", "s3", "memory"] = Field(
        default="file",
        description="Storage backend type",
    )
    path: Optional[str] = Field(
        default=None,
        description="Local file path for file backend",
    )
    bucket: Optional[str] = Field(
        default=None,
        description="S3 bucket name for S3 backend",
    )
    prefix: Optional[str] = Field(
        default=None,
        description="S3 key prefix for organizing objects",
    )
    region: Optional[str] = Field(
        default=None,
        description="AWS region for S3 backend",
    )

    @field_validator("path")
    @classmethod
    def validate_path(cls, v: Optional[str]) -> Optional[str]:
        """Validate and normalize file path."""
        if v is not None:
            v = v.strip()
            if not v:
                return None
        return v

    @field_validator("bucket")
    @classmethod
    def validate_bucket(cls, v: Optional[str]) -> Optional[str]:
        """Validate S3 bucket name."""
        if v is not None:
            v = v.strip()
            if not v:
                return None
            # Basic S3 bucket naming validation
            if len(v) < 3 or len(v) > 63:
                raise ValueError("S3 bucket name must be between 3 and 63 characters")
            if not v[0].isalnum():
                raise ValueError("S3 bucket name must start with a letter or number")
        return v

    def model_post_init(self, __context) -> None:
        """Validate backend-specific requirements."""
        if self.backend == "file" and not self.path:
            raise ValueError("path is required for file backend")
        if self.backend == "s3" and not self.bucket:
            raise ValueError("bucket is required for S3 backend")

validate_path classmethod 

validate_path(v: Optional[str]) -> Optional[str]

Validate and normalize file path.

Source code in src/rotalabs_comply/core/config.py 
185
186
187
188
189
190
191
192
193
@field_validator("path")
@classmethod
def validate_path(cls, v: Optional[str]) -> Optional[str]:
    """Validate and normalize file path."""
    if v is not None:
        v = v.strip()
        if not v:
            return None
    return v

validate_bucket classmethod 

validate_bucket(v: Optional[str]) -> Optional[str]

Validate S3 bucket name.

Source code in src/rotalabs_comply/core/config.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
@field_validator("bucket")
@classmethod
def validate_bucket(cls, v: Optional[str]) -> Optional[str]:
    """Validate S3 bucket name."""
    if v is not None:
        v = v.strip()
        if not v:
            return None
        # Basic S3 bucket naming validation
        if len(v) < 3 or len(v) > 63:
            raise ValueError("S3 bucket name must be between 3 and 63 characters")
        if not v[0].isalnum():
            raise ValueError("S3 bucket name must start with a letter or number")
    return v

model_post_init 

model_post_init(__context) -> None

Validate backend-specific requirements.

Source code in src/rotalabs_comply/core/config.py 
210
211
212
213
214
215
def model_post_init(self, __context) -> None:
    """Validate backend-specific requirements."""
    if self.backend == "file" and not self.path:
        raise ValueError("path is required for file backend")
    if self.backend == "s3" and not self.bucket:
        raise ValueError("bucket is required for S3 backend")

Configuration for storage backend selection.

Attributes:

Attribute Type Default Description
backend Literal["file", "s3", "memory"] "file" Storage backend type
path Optional[str] None Local path for file backend
bucket Optional[str] None S3 bucket for S3 backend
prefix Optional[str] None S3 key prefix
region Optional[str] None AWS region for S3

Example:

from rotalabs_comply import StorageConfig

# File storage
file_config = StorageConfig(
    backend="file",
    path="/var/log/ai-audit/",
)

# S3 storage
s3_config = StorageConfig(
    backend="s3",
    bucket="my-audit-bucket",
    prefix="prod/audit-logs/",
    region="us-west-2",
)

# Memory storage (testing)
memory_config = StorageConfig(backend="memory")

Exceptions

ComplianceError

ComplianceError

Base exception for all compliance-related errors.

All other exceptions in this module inherit from this class, allowing for broad exception catching when needed.

Attributes:

Name Type Description
message

Human-readable error description.
details

Optional dictionary with additional error context.
Source code in src/rotalabs_comply/core/exceptions.py 
12
13
14
15
16
17
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
class ComplianceError(Exception):
    """
    Base exception for all compliance-related errors.

    All other exceptions in this module inherit from this class,
    allowing for broad exception catching when needed.

    Attributes:
        message: Human-readable error description.
        details: Optional dictionary with additional error context.
    """

    def __init__(
        self,
        message: str,
        details: Optional[Dict[str, Any]] = None,
    ) -> None:
        """
        Initialize a ComplianceError.

        Args:
            message: Human-readable error description.
            details: Optional dictionary with additional context about the error.
        """
        self.message = message
        self.details = details or {}
        super().__init__(message)

    def __str__(self) -> str:
        """Return string representation of the error."""
        if self.details:
            return f"{self.message} | Details: {self.details}"
        return self.message

__init__ 

__init__(
    message: str, details: Optional[Dict[str, Any]] = None
) -> None

Initialize a ComplianceError.

Parameters:

Name Type Description Default
message str

Human-readable error description.

 required
details Optional[Dict[str, Any]]

Optional dictionary with additional context about the error.

 None
Source code in src/rotalabs_comply/core/exceptions.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    message: str,
    details: Optional[Dict[str, Any]] = None,
) -> None:
    """
    Initialize a ComplianceError.

    Args:
        message: Human-readable error description.
        details: Optional dictionary with additional context about the error.
    """
    self.message = message
    self.details = details or {}
    super().__init__(message)

__str__ 

__str__() -> str

Return string representation of the error.

Source code in src/rotalabs_comply/core/exceptions.py 
40
41
42
43
44
def __str__(self) -> str:
    """Return string representation of the error."""
    if self.details:
        return f"{self.message} | Details: {self.details}"
    return self.message

Base exception for all compliance-related errors.

Attributes:

Attribute Type Description
message str Human-readable error description
details Dict[str, Any] Additional error context

Example:

from rotalabs_comply import ComplianceError

try:
    # ... compliance operation
    pass
except ComplianceError as e:
    print(f"Error: {e.message}")
    print(f"Details: {e.details}")

AuditError

AuditError

Exception raised for audit logging failures.

This exception is raised when there are issues with: - Writing audit entries - Reading audit logs - Audit log rotation - Audit data validation

Examples:

>>> raise AuditError("Failed to write audit entry", {"entry_id": "abc123"})
Source code in src/rotalabs_comply/core/exceptions.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class AuditError(ComplianceError):
    """
    Exception raised for audit logging failures.

    This exception is raised when there are issues with:
    - Writing audit entries
    - Reading audit logs
    - Audit log rotation
    - Audit data validation

    Examples:
        >>> raise AuditError("Failed to write audit entry", {"entry_id": "abc123"})
    """

    pass

Exception for audit logging failures.

Raised when: - Writing audit entries fails - Reading audit logs fails - Audit log rotation fails - Audit data validation fails

StorageError

StorageError

Exception raised for storage backend failures.

This exception is raised when there are issues with: - Connecting to storage backends (file, S3, etc.) - Reading or writing data to storage - Storage configuration problems - Permission issues

Examples:

>>> raise StorageError("S3 bucket not accessible", {"bucket": "my-bucket"})
Source code in src/rotalabs_comply/core/exceptions.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class StorageError(ComplianceError):
    """
    Exception raised for storage backend failures.

    This exception is raised when there are issues with:
    - Connecting to storage backends (file, S3, etc.)
    - Reading or writing data to storage
    - Storage configuration problems
    - Permission issues

    Examples:
        >>> raise StorageError("S3 bucket not accessible", {"bucket": "my-bucket"})
    """

    pass

Exception for storage backend failures.

Raised when: - Storage backend connection fails - Reading/writing data fails - Storage configuration is invalid - Permission issues occur

EncryptionError

EncryptionError

Exception raised for encryption/decryption failures.

This exception is raised when there are issues with: - Encrypting audit data - Decrypting stored data - Key management - Invalid encryption configuration

Examples:

>>> raise EncryptionError("Invalid encryption key format")
Source code in src/rotalabs_comply/core/exceptions.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class EncryptionError(ComplianceError):
    """
    Exception raised for encryption/decryption failures.

    This exception is raised when there are issues with:
    - Encrypting audit data
    - Decrypting stored data
    - Key management
    - Invalid encryption configuration

    Examples:
        >>> raise EncryptionError("Invalid encryption key format")
    """

    pass

Exception for encryption/decryption failures.

Raised when: - Encrypting audit data fails - Decrypting stored data fails - Key management issues occur - Invalid encryption configuration

ValidationError

ValidationError

Exception raised for data validation failures.

This exception is raised when there are issues with: - Invalid input data format - Missing required fields - Data type mismatches - Schema validation failures

Attributes:

Name Type Description
field

Optional name of the field that failed validation.
value

Optional value that caused the validation failure.

Examples:

>>> raise ValidationError(
...     "Invalid risk level",
...     details={"field": "risk_level", "value": "UNKNOWN"}
... )
Source code in src/rotalabs_comply/core/exceptions.py 
 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
class ValidationError(ComplianceError):
    """
    Exception raised for data validation failures.

    This exception is raised when there are issues with:
    - Invalid input data format
    - Missing required fields
    - Data type mismatches
    - Schema validation failures

    Attributes:
        field: Optional name of the field that failed validation.
        value: Optional value that caused the validation failure.

    Examples:
        >>> raise ValidationError(
        ...     "Invalid risk level",
        ...     details={"field": "risk_level", "value": "UNKNOWN"}
        ... )
    """

    def __init__(
        self,
        message: str,
        details: Optional[Dict[str, Any]] = None,
        field: Optional[str] = None,
        value: Optional[Any] = None,
    ) -> None:
        """
        Initialize a ValidationError.

        Args:
            message: Human-readable error description.
            details: Optional dictionary with additional context.
            field: Optional name of the field that failed validation.
            value: Optional value that caused the validation failure.
        """
        details = details or {}
        if field:
            details["field"] = field
        if value is not None:
            details["value"] = value
        super().__init__(message, details)
        self.field = field
        self.value = value

__init__ 

__init__(
    message: str,
    details: Optional[Dict[str, Any]] = None,
    field: Optional[str] = None,
    value: Optional[Any] = None,
) -> None

Initialize a ValidationError.

Parameters:

Name Type Description Default
message str

Human-readable error description.

 required
details Optional[Dict[str, Any]]

Optional dictionary with additional context.

 None
field Optional[str]

Optional name of the field that failed validation.

 None
value Optional[Any]

Optional value that caused the validation failure.

 None
Source code in src/rotalabs_comply/core/exceptions.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def __init__(
    self,
    message: str,
    details: Optional[Dict[str, Any]] = None,
    field: Optional[str] = None,
    value: Optional[Any] = None,
) -> None:
    """
    Initialize a ValidationError.

    Args:
        message: Human-readable error description.
        details: Optional dictionary with additional context.
        field: Optional name of the field that failed validation.
        value: Optional value that caused the validation failure.
    """
    details = details or {}
    if field:
        details["field"] = field
    if value is not None:
        details["value"] = value
    super().__init__(message, details)
    self.field = field
    self.value = value

Exception for data validation failures.

Additional Attributes:

Attribute Type Description
field Optional[str] Field that failed validation
value Optional[Any] Value that caused failure

Example:

from rotalabs_comply import ValidationError

try:
    # ... validation
    pass
except ValidationError as e:
    print(f"Field: {e.field}")
    print(f"Value: {e.value}")

FrameworkError

FrameworkError

Exception raised for regulatory framework-related failures.

This exception is raised when there are issues with: - Unsupported framework operations - Framework rule validation - Framework configuration errors - Incompatible framework combinations

Attributes:

Name Type Description
framework

Optional identifier of the framework that caused the error.

Examples:

>>> raise FrameworkError(
...     "Framework not supported",
...     details={"framework": "UNKNOWN_FRAMEWORK"}
... )
Source code in src/rotalabs_comply/core/exceptions.py 
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
class FrameworkError(ComplianceError):
    """
    Exception raised for regulatory framework-related failures.

    This exception is raised when there are issues with:
    - Unsupported framework operations
    - Framework rule validation
    - Framework configuration errors
    - Incompatible framework combinations

    Attributes:
        framework: Optional identifier of the framework that caused the error.

    Examples:
        >>> raise FrameworkError(
        ...     "Framework not supported",
        ...     details={"framework": "UNKNOWN_FRAMEWORK"}
        ... )
    """

    def __init__(
        self,
        message: str,
        details: Optional[Dict[str, Any]] = None,
        framework: Optional[str] = None,
    ) -> None:
        """
        Initialize a FrameworkError.

        Args:
            message: Human-readable error description.
            details: Optional dictionary with additional context.
            framework: Optional identifier of the framework that caused the error.
        """
        details = details or {}
        if framework:
            details["framework"] = framework
        super().__init__(message, details)
        self.framework = framework

__init__ 

__init__(
    message: str,
    details: Optional[Dict[str, Any]] = None,
    framework: Optional[str] = None,
) -> None

Initialize a FrameworkError.

Parameters:

Name Type Description Default
message str

Human-readable error description.

 required
details Optional[Dict[str, Any]]

Optional dictionary with additional context.

 None
framework Optional[str]

Optional identifier of the framework that caused the error.

 None
Source code in src/rotalabs_comply/core/exceptions.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def __init__(
    self,
    message: str,
    details: Optional[Dict[str, Any]] = None,
    framework: Optional[str] = None,
) -> None:
    """
    Initialize a FrameworkError.

    Args:
        message: Human-readable error description.
        details: Optional dictionary with additional context.
        framework: Optional identifier of the framework that caused the error.
    """
    details = details or {}
    if framework:
        details["framework"] = framework
    super().__init__(message, details)
    self.framework = framework

Exception for framework-related failures.

Additional Attributes:

Attribute Type Description
framework Optional[str] Framework that caused the error

Raised when: - Unsupported framework operations - Framework rule validation fails - Framework configuration errors - Incompatible framework combinations