All checks were successful
Build and Deploy iObj / build-image (push) Successful in 1h17m41s
- Month-sharded TinyDB storage with search/filter (type, project, tag, date range, free-text) - FastAPI app with bearer-token auth, CRUD + search endpoints - 27 passing pytest tests (storage + API) - Dockerfile + Nomad job spec + Gitea Actions deploy workflow (mirrors devops-mcp pattern)
86 lines
2.4 KiB
Python
86 lines
2.4 KiB
Python
"""Data model for iObj memory objects."""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import date, datetime
|
|
from enum import Enum
|
|
from typing import Any, Optional
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class ObjectType(str, Enum):
|
|
"""The kind of memory object being stored."""
|
|
|
|
GIT_COMMIT = "git_commit"
|
|
SESSION_LOG = "session_log"
|
|
RECAP = "recap"
|
|
MEETING_NOTE = "meeting_note"
|
|
ONEONONE = "oneonone"
|
|
STANDUP = "standup"
|
|
CUSTOM = "custom"
|
|
|
|
|
|
class ObjectCreate(BaseModel):
|
|
"""Payload for creating a new memory object."""
|
|
|
|
type: ObjectType
|
|
object_date: date = Field(
|
|
default_factory=lambda: datetime.now().date(),
|
|
description="The date this object is about (may differ from created_at).",
|
|
)
|
|
project: Optional[str] = None
|
|
tags: list[str] = Field(default_factory=list)
|
|
title: str
|
|
body: str = ""
|
|
payload: dict[str, Any] = Field(default_factory=dict)
|
|
source_machine: Optional[str] = None
|
|
source_tool: Optional[str] = None
|
|
|
|
|
|
class ObjectUpdate(BaseModel):
|
|
"""Payload for partially updating an existing memory object. All fields optional."""
|
|
|
|
type: Optional[ObjectType] = None
|
|
object_date: Optional[date] = None
|
|
project: Optional[str] = None
|
|
tags: Optional[list[str]] = None
|
|
title: Optional[str] = None
|
|
body: Optional[str] = None
|
|
payload: Optional[dict[str, Any]] = None
|
|
source_machine: Optional[str] = None
|
|
source_tool: Optional[str] = None
|
|
|
|
|
|
class MemoryObject(BaseModel):
|
|
"""A fully stored memory object, as returned by the API."""
|
|
|
|
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
|
|
type: ObjectType
|
|
created_at: datetime = Field(default_factory=datetime.now)
|
|
object_date: date
|
|
project: Optional[str] = None
|
|
tags: list[str] = Field(default_factory=list)
|
|
title: str
|
|
body: str = ""
|
|
payload: dict[str, Any] = Field(default_factory=dict)
|
|
source_machine: Optional[str] = None
|
|
source_tool: Optional[str] = None
|
|
|
|
def shard_key(self) -> str:
|
|
"""The YYYY-MM shard this object belongs to, based on object_date."""
|
|
return self.object_date.strftime("%Y-%m")
|
|
|
|
|
|
class SearchQuery(BaseModel):
|
|
"""Search/filter parameters for listing objects."""
|
|
|
|
type: Optional[ObjectType] = None
|
|
project: Optional[str] = None
|
|
tag: Optional[str] = None
|
|
date_from: Optional[date] = None
|
|
date_to: Optional[date] = None
|
|
q: Optional[str] = None
|
|
limit: int = 50
|
|
offset: int = 0
|