51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
"""Abstract persistence API for per-conversation share links."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
# Unambiguous alphabet — no 0/O/1/I/L so codes are easy to read and type.
|
|
_CODE_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
|
|
|
|
|
|
def generate_share_code(length: int = 8) -> str:
|
|
"""Return a random, unguessable, human-readable share code."""
|
|
return "".join(secrets.choice(_CODE_ALPHABET) for _ in range(length))
|
|
|
|
|
|
class ThreadShareStore(ABC):
|
|
@abstractmethod
|
|
async def create(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any]:
|
|
"""Create a new share link for one conversation and return its record.
|
|
|
|
``mode`` is ``"import"`` (recipient copies the conversation) or
|
|
``"view"`` (a public read-only static page).
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def get(self, share_code: str) -> dict[str, Any] | None:
|
|
"""Look up a share by code regardless of owner (None if missing)."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def get_active_for_thread(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any] | None:
|
|
"""Return the owner's non-revoked share of ``mode`` for ``thread_id``, if any.
|
|
|
|
Used so re-sharing the same conversation in the same mode reuses its
|
|
existing code instead of minting a new one each time.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def list_for_owner(self, owner_user_id: str) -> list[dict[str, Any]]:
|
|
"""List all share links created by ``owner_user_id`` (newest first)."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def revoke(self, share_code: str, owner_user_id: str) -> bool:
|
|
"""Revoke a share link. Returns False if missing or not owned."""
|
|
raise NotImplementedError
|