Minimal Guardrails as a Service Implementation

guardrails/api_client.py

@@ -0,0 +1,51 @@
+import os
+from typing import Optional
+
+from guard_rails_api_client import AuthenticatedClient
+from guard_rails_api_client.api.guard import update_guard, validate
+from guard_rails_api_client.models import Guard, ValidatePayload
+from guard_rails_api_client.types import UNSET
+from httpx import Timeout
+
+
+class GuardrailsApiClient:
+    _client: AuthenticatedClient
+    base_url: str
+    api_key: str
+
+    def __init__(self, base_url: Optional[str] = None, api_key: Optional[str] = None):
+        self.base_url = (
+            base_url
+            if base_url is not None
+            else os.environ.get("GUARDRAILS_BASE_URL", "http://localhost:8000")
+        )
+        self.api_key = (
+            api_key if api_key is not None else os.environ.get("GUARDRAILS_API_KEY", "")
+        )
+        self._client = AuthenticatedClient(
+            _base_url=self.base_url,
+            _follow_redirects=True,
+            token=self.api_key,
+            _timeout=Timeout(300),
+        )
+
+    def upsert_guard(self, guard: Guard):
+        update_guard.sync(guard_name=guard.name, client=self._client, body=guard)
+
+    def validate(
+        self,
+        guard: Guard,
+        payload: ValidatePayload,
+        openai_api_key: Optional[str] = None,
+    ):
+        _openai_api_key = (
+            openai_api_key
+            if openai_api_key is not None
+            else os.environ.get("OPENAI_API_KEY", UNSET)
+        )
+        return validate.sync(
+            guard_name=guard.name,
+            client=self._client,
+            body=payload,
+            x_openai_api_key=_openai_api_key,
+        )

guardrails/datatypes.py

@@ -148,6 +148,31 @@ def __eq__(self, other):
             return False
         return self.__dict__ == other.__dict__
 
+    def _to_request(self) -> Dict[str, Any]:
+        element: Dict[str, Any] = {
+            "type": self.tag,
+            "name": self.name,
+            "description": self.description,
+            # This isn't stored anywhere and is inconsistently passed to ValidatorsAttr
+            # (i.e. is never passed to child properties)
+            # meaning its purpose isn't consistenly enforced.
+            # Since this is an XML only property and isn't properly implemented anymore,
+            # I'm just going to ignore it for now.
+            "strict": False,
+            "onFails": [
+                {"validatorTag": v.rail_alias, "method": v.on_fail_descriptor}
+                for v in self.validators_attr.validators
+            ],
+            "dateFormat": getattr(self, "date_format", None),
+            "timeFormat": getattr(self, "time_format", None),
+        }
+        formatters = [v.to_xml_attrib() for v in self.validators_attr.validators]
+        children: Dict[str, Any] = {
+            k: v._to_request() for k, v in self._children.items()
+        }
+
+        return {"children": children, "formatters": formatters, "element": element}
+
 
 registry: Dict[str, Type[DataType]] = {}