Skip to content

Latest commit

 

History

History
151 lines (107 loc) · 3.31 KB

storage.md

File metadata and controls

151 lines (107 loc) · 3.31 KB

Storage

Session storage persists auth tokens across restarts. Used with AuthSession to avoid re-authenticating every time.

How it works

AuthSession.verify() → LoginData → SessionData → storage.save("email", data)
                                                  storage.load("email") → SessionData

SessionData contains everything needed to recreate a session:

  • lid — login token (Authorization / X-Lid header)
  • client_id — stable X-Client-Id UUID
  • session_id — stable X-Session-Id UUID
  • email — email used for login
  • account_id — numeric account ID
  • uri — account URI slug

Backends

MemoryStorage

In-memory dict. Lost on restart. Good for tests.

from teletype.storage import MemoryStorage

storage = MemoryStorage()

FileStorage

One JSON file per session in a directory.

from teletype.storage import FileStorage

storage = FileStorage("./sessions")
# Creates ./sessions/user@example.com.json

SQLiteStorage

Single SQLite database file. Uses aiosqlite.

from teletype.storage import SQLiteStorage

storage = SQLiteStorage("sessions.db")
# Table created automatically on first use

RedisStorage

Redis key-value store. Uses redis-py.

from teletype.storage import RedisStorage

storage = RedisStorage(
    url="redis://localhost:6379",
    prefix="teletype:sessions:",   # key prefix
    ttl=86400 * 30,                # auto-expire after 30 days (optional)
)

MongoStorage

MongoDB collection. Uses motor.

from teletype.storage import MongoStorage

storage = MongoStorage(
    url="mongodb://localhost:27017",
    database="teletype",
    collection="sessions",
)

Using storage with AuthSession

from teletype.auth import AuthSession
from teletype.captcha import TwoCaptchaSolver
from teletype.storage import FileStorage

storage = FileStorage("./sessions")

async with AuthSession(
    solver=TwoCaptchaSolver(api_key="..."),
    storage=storage,
) as auth:
    # Try to restore first
    session = await auth.restore("user@example.com")
    if session:
        print(f"Restored session for @{session.uri}")
        token = session.lid
    else:
        await auth.login("user@example.com")
        login_data = await auth.verify(input("Code: "))
        token = login_data.token

    from teletype import TeletypeClient
    async with TeletypeClient(token) as client:
        account = await client.get_account()

Using storage directly

All backends share the same async interface:

from teletype.storage import FileStorage, SessionData

storage = FileStorage("./sessions")

# Save
await storage.save("my-key", SessionData(lid="token-value"))

# Load
data = await storage.load("my-key")
if data:
    print(data.lid)

# Check existence
exists = await storage.exists("my-key")

# List all keys
all_keys = await storage.keys()

# Delete
await storage.delete("my-key")

Custom backend

Implement BaseStorage:

from teletype.storage import BaseStorage, SessionData

class MyStorage(BaseStorage):
    async def save(self, key: str, data: SessionData) -> None: ...
    async def load(self, key: str) -> SessionData | None: ...
    async def delete(self, key: str) -> None: ...
    async def exists(self, key: str) -> bool: ...
    async def keys(self) -> list[str]: ...