Viking URI handler. URI Format: viking:// / Scopes: - resources: Independent resource scope (viking://resources/{project}/...) - user: User scope (viking://user/...), including sessions under viking://user/{user_id}/sessions/{session_id} - session: Legacy
| 12 | |
| 13 | |
| 14 | class VikingURI: |
| 15 | """ |
| 16 | Viking URI handler. |
| 17 | |
| 18 | URI Format: viking://<scope>/<path> |
| 19 | |
| 20 | Scopes: |
| 21 | - resources: Independent resource scope (viking://resources/{project}/...) |
| 22 | - user: User scope (viking://user/...), including sessions under |
| 23 | viking://user/{user_id}/sessions/{session_id} |
| 24 | - session: Legacy alias for user sessions (viking://session/{session_id}/...) |
| 25 | - agent: Agent capabilities scope (viking://agent/skills/..., viking://agent/endpoints/...) |
| 26 | - queue: Queue scope (viking://queue/...) |
| 27 | |
| 28 | Examples: |
| 29 | - viking://resources/my_project/docs/api |
| 30 | - viking://user/memories/preferences/code_style |
| 31 | - viking://user/skills/pdf |
| 32 | - viking://user/alice/sessions/session123/messages.jsonl |
| 33 | """ |
| 34 | |
| 35 | SCHEME = "viking" |
| 36 | # SCOPES that can be listed in root directory (ov ls) |
| 37 | LISTABLE_SCOPES = { |
| 38 | "resources", |
| 39 | "user", |
| 40 | "agent", |
| 41 | } |
| 42 | PUBLIC_SCOPES = frozenset(LISTABLE_SCOPES) |
| 43 | LEGACY_SCOPES = frozenset({"session"}) |
| 44 | INTERNAL_SCOPES = frozenset({"temp", "queue", "upload"}) |
| 45 | # All valid scopes that can be addressed by the URI parser/storage internals. |
| 46 | # Public API handlers must not use this as their external whitelist. |
| 47 | VISITABLE_SCOPES = PUBLIC_SCOPES | LEGACY_SCOPES | INTERNAL_SCOPES |
| 48 | |
| 49 | def __init__(self, uri: str): |
| 50 | """ |
| 51 | Initialize URI handler. |
| 52 | |
| 53 | Accepts both full-format (viking://...) and short-format (/resources, resources) |
| 54 | URIs. Short-format URIs are automatically normalized to full format. |
| 55 | |
| 56 | Args: |
| 57 | uri: URI string (full or short format) |
| 58 | """ |
| 59 | self.uri = self.normalize(uri) |
| 60 | self._parsed = self._parse() |
| 61 | |
| 62 | def _parse(self) -> Dict[str, str]: |
| 63 | """ |
| 64 | Parse Viking URI into components. |
| 65 | |
| 66 | Returns: |
| 67 | Dictionary with URI components |
| 68 | """ |
| 69 | if not self.uri.startswith(f"{self.SCHEME}://"): |
| 70 | raise ValueError(f"URI must start with '{self.SCHEME}://'") |
| 71 |
no outgoing calls