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 | None, Depends(get_redis_client)], 

35): 

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

37 try: 

38 if redis_client: 

39 await redis_client.delete(cache_key) 

40 except RedisError as e: 

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

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

43 

44 return {"status": "success"} 

45 

46 

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

48async def get_file( 

49 site_id: SiteId, 

50 path: SafePath, 

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

52): 

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

54 path = path + "index.html" 

55 

56 try: 

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

58 except AWSStorageError: 

59 return Response(status_code=404) 

60 

61 media_type, _ = mimetypes.guess_type(path) 

62 return Response( 

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

64 ) 

65 

66 

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

68async def create_upload_session( 

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

70) -> AWSUploadSessionResponse: 

71 session = await aws_storage.create_upload_session() 

72 return AWSUploadSessionResponse( 

73 site_id=session.site_id, 

74 bucket=session.bucket, 

75 region=session.region, 

76 access_key_id=session.access_key_id, 

77 secret_access_key=session.secret_access_key, 

78 session_token=session.session_token, 

79 )