Build conversation ownership + RBAC sharing

Loading diagram...

• Line 1: Declares a Mermaid ER (Entity-Relationship) diagram block, indicating the code defines a database schema visualization.
• Line 2: Defines a one-to-many relationship where one USER creates zero or more CONVERSATION records.
• Line 3: Defines a one-to-many relationship where one USER is granted access to zero or more CONVERSATION_ACL (access control list) entries.
• Lines 34-43: Defines the SHARE_LINK entity for shareable conversation links, with a UUID primary key, a conversation_id foreign key, a unique token string used in the URL, an access_level enum controlling what the link grants, a created_by foreign key, max_uses and current_uses integers for limiting link redemptions, and an expires_at timestamp for link expiration.

1import enum
2import secrets
3from datetime import datetime, timezone
4from uuid import uuid4
5
6from sqlalchemy import (
7    Column, DateTime, Enum, ForeignKey, Integer,
8    String, UniqueConstraint, CheckConstraint,
9)
10from sqlalchemy.dialects.postgresql import UUID
11from sqlalchemy.orm import relationship, validates
12from app.database import Base
13
14class AccessLevel(enum.Enum):
15    OWNER = "owner"
16    EDITOR = "editor"
17    VIEWER = "viewer"
18
19ACCESS_HIERARCHY = {
20    AccessLevel.OWNER: 3,
21    AccessLevel.EDITOR: 2,
22    AccessLevel.VIEWER: 1,
23}
24
25class ConversationACL(Base):
26    __tablename__ = "conversation_acl"
27    __table_args__ = (
28        UniqueConstraint("conversation_id", "user_id", name="uq_conv_user"),
29        CheckConstraint(
30            "access_level IN ('owner', 'editor', 'viewer')",
31            name="ck_access_level",
32        ),
33    )
34
35    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
36    conversation_id = Column(
37        UUID(as_uuid=True), ForeignKey("conversations.id", ondelete="CASCADE"),
38        nullable=False, index=True,
39    )
40    user_id = Column(
41        UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"),
42        nullable=False, index=True,
43    )
44    access_level = Column(Enum(AccessLevel), nullable=False)
45    granted_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
46    granted_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
47    expires_at = Column(DateTime(timezone=True), nullable=True)
48
49    conversation = relationship("Conversation", back_populates="acl_entries")
50    user = relationship("User", foreign_keys=[user_id])
51    granter = relationship("User", foreign_keys=[granted_by])
52
53    @validates("access_level")
54    def validate_access_level(self, key, value):
55        if isinstance(value, str):
56            return AccessLevel(value)
57        return value
58
59    def is_expired(self) -> bool:
60        if self.expires_at is None:
61            return False
62        return datetime.now(timezone.utc) > self.expires_at
63
64    def has_minimum_access(self, required: AccessLevel) -> bool:
65        if self.is_expired():
66            return False
67        return ACCESS_HIERARCHY[self.access_level] >= ACCESS_HIERARCHY[required]
68
69class ShareLink(Base):
70    __tablename__ = "share_links"
71
72    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
73    conversation_id = Column(
74        UUID(as_uuid=True), ForeignKey("conversations.id", ondelete="CASCADE"),
75        nullable=False,
76    )
77    token = Column(String(64), unique=True, default=lambda: secrets.token_urlsafe(32))
78    access_level = Column(Enum(AccessLevel), nullable=False, default=AccessLevel.VIEWER)
79    created_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
80    max_uses = Column(Integer, nullable=True)
81    current_uses = Column(Integer, default=0)
82    expires_at = Column(DateTime(timezone=True), nullable=True)

• Lines 1-6: Import the standard library modules. The secrets module provides token_urlsafe which generates cryptographically secure tokens suitable for share URLs. The enum module defines the access level as a Python enum rather than raw strings, preventing typos and invalid states.
• Lines 8-12: Import SQLAlchemy column types and constraints. The UUID type from the PostgreSQL dialect stores conversation and user identifiers as native PostgreSQL UUIDs rather than strings, giving you indexed 128-bit keys with no collation overhead.
• Lines 15-18: Define AccessLevel as a Python enum.Enum with three string values. Using an enum ensures that the database column, Python models, and API schemas all share a single source of truth for valid access levels.
• Lines 69-82: The ShareLink model stores link-based sharing tokens. The token column defaults to secrets.token_urlsafe(32), producing a 43-character URL-safe string with 256 bits of entropy. The max_uses column enables single-use or limited-use links—when None, the link has unlimited redemptions.

1from fastapi import APIRouter, Depends, HTTPException, status
2from pydantic import BaseModel, Field
3from sqlalchemy import select
4from sqlalchemy.ext.asyncio import AsyncSession
5from typing import Optional
6from uuid import UUID
7
8from app.auth import get_current_user
9from app.database import get_db
10from app.redis_client import get_redis
11from app.models import AccessLevel, ConversationACL, ShareLink
12
13router = APIRouter(prefix="/conversations", tags=["sharing"])
14
15class ShareRequest(BaseModel):
16    access_level: str = Field(default="viewer", pattern="^(editor|viewer)$")
17    max_uses: Optional[int] = Field(default=None, ge=1, le=1000)
18    expires_in_hours: Optional[int] = Field(default=None, ge=1, le=720)
19
20class ShareResponse(BaseModel):
21    share_url: str
22    access_level: str
23    max_uses: Optional[int]
24    expires_at: Optional[str]
25
26def require_access(minimum: AccessLevel):
27    async def dependency(
28        conversation_id: UUID,
29        db: AsyncSession = Depends(get_db),
30        user: dict = Depends(get_current_user),
31    ):
32        result = await db.execute(
33            select(ConversationACL).where(
34                ConversationACL.conversation_id == conversation_id,
35                ConversationACL.user_id == user["sub"],
36            )
37        )
38        acl_entry = result.scalar_one_or_none()
39        if acl_entry is None or not acl_entry.has_minimum_access(minimum):
40            raise HTTPException(
41                status_code=status.HTTP_403_FORBIDDEN,
42                detail=f"Requires {minimum.value} access",
43            )
44        return acl_entry
45    return dependency
46
47@router.post("/{conversation_id}/share", response_model=ShareResponse)
48async def create_share(
49    conversation_id: UUID,
50    body: ShareRequest,
51    acl: ConversationACL = Depends(require_access(AccessLevel.OWNER)),
52    db: AsyncSession = Depends(get_db),
53    redis=Depends(get_redis),
54):
55    expires_at = None
56    if body.expires_in_hours:
57        from datetime import datetime, timedelta, timezone
58        expires_at = datetime.now(timezone.utc) + timedelta(hours=body.expires_in_hours)
59
60    link = ShareLink(
61        conversation_id=conversation_id,
62        access_level=AccessLevel(body.access_level),
63        created_by=acl.user_id,
64        max_uses=body.max_uses,
65        expires_at=expires_at,
66    )
67    db.add(link)
68    await db.flush()
69
70    cache_key = f"share:{link.token}"
71    cache_value = f"{conversation_id}:{body.access_level}"
72    ttl = body.expires_in_hours * 3600 if body.expires_in_hours else 86400 * 30
73    await redis.setex(cache_key, ttl, cache_value)
74
75    await db.commit()
76    share_url = f"https://app.example.com/share/{link.token}"
77    return ShareResponse(
78        share_url=share_url,
79        access_level=body.access_level,
80        max_uses=body.max_uses,
81        expires_at=expires_at.isoformat() if expires_at else None,
82    )

• Lines 1-11: Import FastAPI components alongside the custom authentication dependency get_current_user, which extracts and validates the JWT from the Authorization header using PyJWT. The get_redis dependency returns the same Redis connection pool used by the session management layer, ensuring share link caching shares connection resources.
• Lines 16-19: The ShareRequest Pydantic model constrains the access_level field with a regex pattern that only allows "editor" or "viewer"—owners cannot create share links that grant owner-level access, preventing privilege escalation through sharing. The max_uses field is bounded between 1 and 1000 to prevent abuse.
• Lines 29-48: The require_access factory function returns an async dependency closure. When FastAPI resolves this dependency, it queries the conversation_acl table for the requesting user's entry on the target conversation. If no entry exists or the entry's access level is below the required minimum, a 403 Forbidden response is returned immediately—the route handler never executes. This pattern centralizes authorization logic so that adding a new endpoint only requires Depends(require_access(AccessLevel.EDITOR)) in the function signature.
• Lines 78-85: Only after the Redis write succeeds does the transaction commit. The response returns the full share URL, access level, and expiry metadata. In a production system, the URL domain would come from an environment variable rather than a hardcoded string.