Coverage for app / api / deps.py: 100%

1from typing import Optional 

2from fastapi import Depends, HTTPException, status, Header 

3from sqlalchemy.ext.asyncio import AsyncSession 

4from sqlalchemy import select 

5 

6from app.db.database import get_db 

7from app.db.models import Agent, Creator 

8from app.core.security import verify_api_key, decode_access_token 

9 

10 

11async def get_current_agent( 

12 x_agent_api_key: Optional[str] = Header(None), 

13 db: AsyncSession = Depends(get_db) 

14) -> Optional[Agent]: 

15 """Get the current agent from API key header. Returns None if no key provided.""" 

16 if not x_agent_api_key: 

17 return None 

18 

19 # We need to check all agents since we hash the key 

20 result = await db.execute(select(Agent)) 

21 agents = result.scalars().all() 

22 

23 for agent in agents: 

24 if verify_api_key(x_agent_api_key, agent.api_key_hash): 

25 return agent 

26 

27 return None 

28 

29 

30async def get_required_agent( 

31 agent: Optional[Agent] = Depends(get_current_agent) 

32) -> Agent: 

33 """Require a valid agent API key.""" 

34 if not agent: 

35 raise HTTPException( 

36 status_code=status.HTTP_401_UNAUTHORIZED, 

37 detail="Invalid or missing API key" 

38 ) 

39 return agent 

40 

41 

42async def get_current_creator( 

43 authorization: Optional[str] = Header(None), 

44 db: AsyncSession = Depends(get_db) 

45) -> Optional[Creator]: 

46 """Get the current creator from JWT token.""" 

47 if not authorization or not authorization.startswith("Bearer "): 

48 return None 

49 

50 token = authorization.split(" ")[1] 

51 payload = decode_access_token(token) 

52 

53 if not payload or "sub" not in payload: 

54 return None 

55 

56 creator_id = payload["sub"] 

57 result = await db.execute(select(Creator).where(Creator.id == creator_id)) 

58 return result.scalar_one_or_none() 

59 

60 

61async def get_required_creator( 

62 creator: Optional[Creator] = Depends(get_current_creator) 

63) -> Creator: 

64 """Require a valid creator JWT token.""" 

65 if not creator: 

66 raise HTTPException( 

67 status_code=status.HTTP_401_UNAUTHORIZED, 

68 detail="Invalid or missing authentication" 

69 ) 

70 return creator