File size: 2,993 Bytes
1fb4b54 f7959b4 1fb4b54 f7959b4 1fb4b54 | 1 2 3 4 5 6 7 8 9 10 11 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 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 | from pydantic import BaseModel, Field, field_validator
from typing import Optional, Literal, List, Any, Dict
from enum import Enum
# ---------------------------------------------------------------------------
# Fallback enums – used when the proprietary core engine is not installed.
# These mirror the canonical definitions from the public specification.
# ---------------------------------------------------------------------------
class ResourceType(str, Enum):
DATABASE = "database"
STORAGE_ACCOUNT = "storage_account"
VM = "vm"
VIRTUAL_NETWORK = "virtual_network"
# enterprise-only types omitted for public sandbox
class PermissionLevel(str, Enum):
READ = "read"
WRITE = "write"
ADMIN = "admin"
class Environment(str, Enum):
DEV = "dev"
STAGING = "staging"
PROD = "prod"
class ChangeScope(str, Enum):
MINOR = "minor"
MAJOR = "major"
CRITICAL = "critical"
# ---------------------------------------------------------------------------
# Optional import from protected core engine – not available in public Spaces
try:
from agentic_reliability_framework.core.governance.intents import (
ResourceType,
PermissionLevel,
Environment,
ChangeScope,
)
except ImportError:
# The fallback enums defined above are used.
pass
class BaseIntentRequest(BaseModel):
environment: Environment
estimated_cost: Optional[float] = Field(None, ge=0)
policy_violations: List[str] = Field(default_factory=list)
requester: str = Field(...)
provenance: Dict[str, Any] = Field(default_factory=dict)
class ProvisionResourceRequest(BaseIntentRequest):
intent_type: Literal["provision_resource"] = "provision_resource"
resource_type: ResourceType
region: str
size: str
configuration: Dict[str, Any] = Field(default_factory=dict)
@field_validator("region")
def validate_region(cls, v):
return v
class GrantAccessRequest(BaseIntentRequest):
intent_type: Literal["grant_access"] = "grant_access"
principal: str
permission_level: PermissionLevel
resource_scope: str
justification: Optional[str] = None
@field_validator("resource_scope")
def validate_resource_scope(cls, v):
if not v.startswith("/"):
raise ValueError("Resource scope must start with '/'")
return v
class DeployConfigurationRequest(BaseIntentRequest):
intent_type: Literal["deploy_config"] = "deploy_config"
service_name: str
change_scope: ChangeScope
deployment_target: Environment
risk_level_hint: Optional[float] = Field(None, ge=0, le=1)
configuration: Dict[str, Any] = Field(default_factory=dict)
@field_validator("service_name")
def validate_service_name(cls, v):
if len(v) < 3:
raise ValueError("Service name must be at least 3 characters")
return v
InfrastructureIntentRequest = ProvisionResourceRequest | GrantAccessRequest | DeployConfigurationRequest |