Draft Storage¶

Хранение черновиков и незавершённых проектов пользователей.

Возможности¶

Автосохранение черновиков
Версионирование
TTL для устаревших черновиков
Поиск и фильтрация
Export/Import

Использование¶

Базовое использование¶

from kit.services import DraftStorage

storage = DraftStorage(storage_path="./drafts")

# Сохранение черновика
await storage.save(
    user_id="user_123",
    project_id="project_1",
    data={
        "title": "Мой проект",
        "content": "...",
        "settings": {"style": "cinematic"}
    }
)

# Загрузка
draft = await storage.load("user_123", "project_1")
print(draft["title"])  # "Мой проект"

# Удаление
await storage.delete("user_123", "project_1")

Автосохранение¶

storage = DraftStorage(
    storage_path="./drafts",
    auto_save_interval=30  # Автосохранение каждые 30 секунд
)

# Регистрация проекта для автосохранения
storage.register_autosave("user_123", "project_1", get_data_func)

# Отмена автосохранения
storage.unregister_autosave("user_123", "project_1")

Версионирование¶

storage = DraftStorage(
    storage_path="./drafts",
    max_versions=10  # Хранить 10 последних версий
)

# Сохранение создаёт новую версию
await storage.save("user_123", "project_1", {"v": 1})
await storage.save("user_123", "project_1", {"v": 2})
await storage.save("user_123", "project_1", {"v": 3})

# Получение истории версий
versions = await storage.get_versions("user_123", "project_1")
# [{"version": 1, "timestamp": ...}, {"version": 2, ...}, ...]

# Загрузка конкретной версии
old_draft = await storage.load("user_123", "project_1", version=1)

Список черновиков пользователя¶

# Все черновики пользователя
drafts = await storage.list_user_drafts("user_123")

for draft in drafts:
    print(f"{draft['project_id']}: {draft['title']} (updated: {draft['updated_at']})")

# С фильтрацией
recent = await storage.list_user_drafts(
    "user_123",
    filter_fn=lambda d: d["updated_at"] > yesterday
)

TTL для черновиков¶

storage = DraftStorage(
    storage_path="./drafts",
    default_ttl=604800  # 7 дней
)

# Черновики автоматически удаляются через 7 дней
await storage.save("user_123", "temp_project", data)

# Кастомный TTL
await storage.save("user_123", "important", data, ttl=2592000)  # 30 дней

Export/Import¶

# Export всех черновиков пользователя
export_data = await storage.export_user_drafts("user_123")
with open("backup.json", "w") as f:
    json.dump(export_data, f)

# Import
with open("backup.json") as f:
    import_data = json.load(f)
await storage.import_user_drafts("user_123", import_data)

API Reference¶

DraftStorage¶

class DraftStorage:
    def __init__(
        self,
        storage_path: str = "./drafts",
        max_versions: int = 5,
        default_ttl: int = None,
        auto_save_interval: int = None
    )

    async def save(
        self,
        user_id: str,
        project_id: str,
        data: dict,
        ttl: int = None
    ) -> None

    async def load(
        self,
        user_id: str,
        project_id: str,
        version: int = None  # None = latest
    ) -> Optional[dict]

    async def delete(
        self,
        user_id: str,
        project_id: str
    ) -> bool

    async def exists(
        self,
        user_id: str,
        project_id: str
    ) -> bool

    async def list_user_drafts(
        self,
        user_id: str,
        filter_fn: Callable = None
    ) -> List[DraftInfo]

    async def get_versions(
        self,
        user_id: str,
        project_id: str
    ) -> List[VersionInfo]

    async def export_user_drafts(self, user_id: str) -> dict
    async def import_user_drafts(self, user_id: str, data: dict) -> int

    def register_autosave(
        self,
        user_id: str,
        project_id: str,
        get_data: Callable
    ) -> None
    def unregister_autosave(self, user_id: str, project_id: str) -> None

    async def cleanup_expired(self) -> int

DraftInfo¶

@dataclass
class DraftInfo:
    project_id: str
    title: str
    created_at: datetime
    updated_at: datetime
    version: int
    size_bytes: int

@dataclass
class VersionInfo:
    version: int
    timestamp: datetime
    size_bytes: int

Примеры из production¶

Autoshorts — черновики видео-проектов¶

class VideoProjectStorage:
    def __init__(self):
        self.storage = DraftStorage(
            storage_path="./video_drafts",
            max_versions=10,
            default_ttl=604800 * 2  # 2 недели
        )

    async def save_project(self, user_id: str, project: dict) -> str:
        """Сохранение проекта."""
        project_id = project.get("id") or str(uuid4())

        await self.storage.save(user_id, project_id, {
            "id": project_id,
            "title": project["title"],
            "script": project["script"],
            "settings": project["settings"],
            "assets": project.get("assets", []),
            "status": project.get("status", "draft")
        })

        return project_id

    async def get_user_projects(self, user_id: str) -> List[dict]:
        """Список проектов пользователя."""
        drafts = await self.storage.list_user_drafts(user_id)

        projects = []
        for draft in drafts:
            data = await self.storage.load(user_id, draft.project_id)
            projects.append({
                "id": draft.project_id,
                "title": data.get("title", "Untitled"),
                "status": data.get("status", "draft"),
                "updated_at": draft.updated_at
            })

        return sorted(projects, key=lambda x: x["updated_at"], reverse=True)

    async def restore_version(
        self,
        user_id: str,
        project_id: str,
        version: int
    ) -> dict:
        """Восстановление версии проекта."""
        old_data = await self.storage.load(user_id, project_id, version=version)

        if old_data:
            await self.storage.save(user_id, project_id, old_data)

        return old_data

Dashboard — черновики настроек¶

class SettingsDraftStorage:
    def __init__(self):
        self.storage = DraftStorage(
            storage_path="./settings_drafts",
            default_ttl=3600  # 1 час
        )

    async def save_draft(self, user_id: str, settings: dict):
        """Автосохранение настроек."""
        await self.storage.save(
            user_id,
            "settings_draft",
            {
                "settings": settings,
                "saved_at": datetime.now().isoformat()
            }
        )

    async def get_draft(self, user_id: str) -> Optional[dict]:
        """Получение черновика настроек."""
        draft = await self.storage.load(user_id, "settings_draft")
        return draft.get("settings") if draft else None

    async def discard_draft(self, user_id: str):
        """Отмена черновика."""
        await self.storage.delete(user_id, "settings_draft")

Real-time collaboration¶

class CollaborativeDraftStorage:
    def __init__(self):
        self.storage = DraftStorage(
            storage_path="./collab_drafts",
            max_versions=50,  # Больше версий для истории
            auto_save_interval=5  # Частое автосохранение
        )
        self.locks = {}  # project_id -> user_id

    async def acquire_lock(self, project_id: str, user_id: str) -> bool:
        """Блокировка проекта для редактирования."""
        if project_id in self.locks and self.locks[project_id] != user_id:
            return False
        self.locks[project_id] = user_id
        return True

    async def release_lock(self, project_id: str, user_id: str):
        """Снятие блокировки."""
        if self.locks.get(project_id) == user_id:
            del self.locks[project_id]

    async def save_with_conflict_check(
        self,
        user_id: str,
        project_id: str,
        data: dict,
        expected_version: int
    ) -> bool:
        """Сохранение с проверкой конфликтов."""
        versions = await self.storage.get_versions(user_id, project_id)
        current_version = versions[-1].version if versions else 0

        if current_version != expected_version:
            return False  # Конфликт версий

        await self.storage.save(user_id, project_id, data)
        return True