Coverage for /usr/lib/python3.10/site-packages/hyd/backend/mount_helper.py: 48%

1from pathlib import Path 

2 

3from fastapi.staticfiles import StaticFiles 

4from sqlalchemy.orm import Session 

5from starlette.routing import BaseRoute, Router 

6 

7import hyd.backend.project.service as project_service 

8from hyd.backend.tag.models import TagEntry 

9from hyd.backend.util.const import PATH_PROJECTS 

10from hyd.backend.util.logger import HydLogger 

11from hyd.backend.util.models import NameStr, PrimaryKey 

12from hyd.backend.version.models import VersionEntry 

13 

14LOGGER = HydLogger("MountHelper") 

15 

16#################################################################################################### 

17#### MountHelper 

18#################################################################################################### 

19 

20 

21class MountHelper: 

22 router: Router 

23 

24 url_route_mapping: dict[str, tuple[Path, BaseRoute]] = {} 

25 

26 @classmethod 

27 def set_router(cls, router: Router) -> None: 

28 cls.router = router 

29 

30 @classmethod 

31 def mount_version(cls, version_entry: VersionEntry) -> None: 

32 id = version_entry.project_id 

33 name = version_entry.project_entry.name 

34 vers = version_entry.version 

35 

36 relative_url = _relative_version_url(name, vers) 

37 path = path_to_version(id, vers) 

38 

39 cls.router.mount(relative_url, StaticFiles(directory=path, html=True)) 

40 cls.url_route_mapping[relative_url] = (path, cls.router.routes[-1]) 

41 

42 LOGGER.debug("%s --> %s", relative_url, path) 

43 

44 @classmethod 

45 def unmount_version(cls, project_name: NameStr, version: NameStr) -> None: 

46 relative_url = _relative_version_url(project_name, version) 

47 

48 path, route = cls.url_route_mapping[relative_url] 

49 cls.router.routes.remove(route) 

50 

51 LOGGER.debug("%s -x- %s", relative_url, path) 

52 

53 @classmethod 

54 def mount_tag(cls, tag_entry: TagEntry) -> None: 

55 id = tag_entry.project_id 

56 tag = tag_entry.tag 

57 name = tag_entry.project_entry.name 

58 version = tag_entry.version 

59 

60 relative_url = _relative_tag_url(name, tag) 

61 path = path_to_version(id, version) 

62 

63 cls.router.mount(relative_url, StaticFiles(directory=path, html=True)) 

64 cls.url_route_mapping[relative_url] = (path, cls.router.routes[-1]) 

65 

66 LOGGER.debug("%s --> %s", relative_url, path) 

67 

68 @classmethod 

69 def unmount_tag(cls, project_name: NameStr, tag: NameStr) -> None: 

70 relative_url = _relative_tag_url(project_name, tag) 

71 

72 path, route = cls.url_route_mapping[relative_url] 

73 cls.router.routes.remove(route) 

74 

75 LOGGER.debug("%s -x- %s", relative_url, path) 

76 

77 @classmethod 

78 def mount_existing_projects(cls, *, db: Session) -> None: 

79 project_entries = project_service.read_projects(db=db) 

80 

81 for project_entry in project_entries: 

82 

83 version_entries: list[VersionEntry] = project_entry.version_entries 

84 for version_entry in version_entries: 

85 cls.mount_version(version_entry=version_entry) 

86 

87 tag_entries: list[TagEntry] = project_entry.tag_entries 

88 for tag_entry in tag_entries: 

89 if tag_entry.version: 

90 cls.mount_tag(tag_entry=tag_entry) 

91 

92 

93#################################################################################################### 

94#### Util 

95#################################################################################################### 

96 

97 

98def path_to_version(project_id: PrimaryKey, version: NameStr) -> Path: 

99 return PATH_PROJECTS / str(project_id) / version 

100 

101 

102def _relative_version_url(name: str, version: str) -> str: 

103 return f"/project/{name}/v/{version}" 

104 

105 

106def _relative_tag_url(name: str, tag: str) -> str: 

107 return f"/project/{name}/t/{tag}"