Coverage for backend / app / routers / coverage.py: 100%

1import mimetypes 

2 

3import re 

4from typing import Annotated 

5 

6from fastapi import APIRouter, Depends, HTTPException, Path 

7from fastapi.responses import Response 

8from redis import RedisError 

9from redis.asyncio import Redis 

10 

11from app.dependencies.redis_client import get_redis_client 

12from app.dependencies.auth import verify_api_key 

13 

14from app.utils.aws_storage import AWSStorage, AWSStorageError 

15from app.dependencies.aws_storage import get_aws_storage 

16from app.constants import BADGE_CACHE_KEY 

17from app.schemas import AWSUploadSessionResponse 

18 

19SAFE_PATH_RE = re.compile(r"^(?!.*(?:^|/)\.\.(?:/|$)).*$") 

20 

21SiteId = Annotated[str, Path(pattern=r"^[a-f0-9]{12}$")] 

22SafePath = Annotated[str, Path(pattern=SAFE_PATH_RE)] # ty: ignore[invalid-argument-type] 

23 

24router = APIRouter() 

25 

26 

27@router.post( 

28 "/invalidate-cache/{repo_owner}/{repo_name}/", 

29 dependencies=[Depends(verify_api_key)], 

30) 

31async def invalidate_cache( 

32 repo_owner: str, 

33 repo_name: str, 

34 redis_client: Annotated[Redis, Depends(get_redis_client)], 

35): 

36 cache_key = BADGE_CACHE_KEY.format(org=repo_owner, repo=repo_name) 

37 try: 

38 await redis_client.delete(cache_key) 

39 except RedisError as e: 

40 print(f"Error invalidating cache for {repo_owner}/{repo_name}: {e}") 

41 raise HTTPException(status_code=500, detail="Failed to invalidate cache") 

42 

43 return {"status": "success"} 

44 

45 

46@router.get("/{site_id}/{path:path}") 

47async def get_file( 

48 site_id: SiteId, 

49 path: SafePath, 

50 aws_storage: Annotated[AWSStorage, Depends(get_aws_storage)], 

51): 

52 if path == "" or path.endswith("/"): 

53 path = path + "index.html" 

54 

55 try: 

56 content = await aws_storage.get_file(site_id, path) 

57 except AWSStorageError: 

58 return Response(status_code=404) 

59 

60 media_type, _ = mimetypes.guess_type(path) 

61 return Response( 

62 content=content, media_type=media_type or "application/octet-stream" 

63 ) 

64 

65 

66@router.post("/create-site/", dependencies=[Depends(verify_api_key)]) 

67async def create_upload_session( 

68 aws_storage: Annotated[AWSStorage, Depends(get_aws_storage)], 

69) -> AWSUploadSessionResponse: 

70 session = await aws_storage.create_upload_session() 

71 return AWSUploadSessionResponse( 

72 site_id=session.site_id, 

73 bucket=session.bucket, 

74 region=session.region, 

75 access_key_id=session.access_key_id, 

76 secret_access_key=session.secret_access_key, 

77 session_token=session.session_token, 

78 )