chore: fix linting

This commit is contained in:
Anish Sarkar 2026-06-25 04:31:36 +05:30
parent d6bffa6f07
commit 2e33ba7723
3 changed files with 11 additions and 9 deletions

View file

@ -1,7 +1,7 @@
"""Zero sync authentication context routes.""" """Zero sync authentication context routes."""
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends
from pydantic import BaseModel from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.auth.context import AuthContext from app.auth.context import AuthContext
@ -13,8 +13,10 @@ router = APIRouter(prefix="/zero", tags=["zero"])
class ZeroContextResponse(BaseModel): class ZeroContextResponse(BaseModel):
userId: str model_config = ConfigDict(populate_by_name=True)
allowedSpaceIds: list[int]
user_id: str = Field(alias="userId")
allowed_space_ids: list[int] = Field(alias="allowedSpaceIds")
@router.get("/context", response_model=ZeroContextResponse) @router.get("/context", response_model=ZeroContextResponse)
@ -24,6 +26,6 @@ async def get_zero_context(
) -> ZeroContextResponse: ) -> ZeroContextResponse:
allowed_space_ids = await get_allowed_read_space_ids(session, auth) allowed_space_ids = await get_allowed_read_space_ids(session, auth)
return ZeroContextResponse( return ZeroContextResponse(
userId=str(auth.user.id), user_id=str(auth.user.id),
allowedSpaceIds=allowed_space_ids, allowed_space_ids=allowed_space_ids,
) )

View file

@ -435,7 +435,6 @@ async def list_snapshots_for_thread(
thread_id: int, thread_id: int,
auth: AuthContext, auth: AuthContext,
) -> list[dict]: ) -> list[dict]:
user = auth.user
"""List all public snapshots for a thread.""" """List all public snapshots for a thread."""
from app.config import config from app.config import config
@ -482,7 +481,6 @@ async def list_snapshots_for_search_space(
search_space_id: int, search_space_id: int,
auth: AuthContext, auth: AuthContext,
) -> list[dict]: ) -> list[dict]:
user = auth.user
"""List all public snapshots for a search space.""" """List all public snapshots for a search space."""
from app.config import config from app.config import config
@ -540,7 +538,6 @@ async def delete_snapshot(
snapshot_id: int, snapshot_id: int,
auth: AuthContext, auth: AuthContext,
) -> bool: ) -> bool:
user = auth.user
"""Delete a specific snapshot. Only thread owner can delete.""" """Delete a specific snapshot. Only thread owner can delete."""
# Get snapshot with thread # Get snapshot with thread
result = await session.execute( result = await session.execute(

View file

@ -18,6 +18,7 @@ logger = logging.getLogger(__name__)
PAT_PREFIX = "ss_pat_" PAT_PREFIX = "ss_pat_"
PAT_TOKEN_BYTES = 32 PAT_TOKEN_BYTES = 32
LAST_USED_THROTTLE = timedelta(minutes=10) LAST_USED_THROTTLE = timedelta(minutes=10)
_last_used_tasks: set[asyncio.Task[None]] = set()
def generate_pat() -> str: def generate_pat() -> str:
@ -70,4 +71,6 @@ def maybe_touch_last_used(pat: PersonalAccessToken) -> None:
if last_used_at is not None and now - last_used_at < LAST_USED_THROTTLE: if last_used_at is not None and now - last_used_at < LAST_USED_THROTTLE:
return return
asyncio.create_task(_touch_last_used(pat.id)) task = asyncio.create_task(_touch_last_used(pat.id))
_last_used_tasks.add(task)
task.add_done_callback(_last_used_tasks.discard)