Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/events/filesystems.py: 86%

1from dataclasses import dataclass 

2from datetime import datetime, timedelta 

3from enum import Enum 

4from pathlib import Path 

5 

6from coxbuild import get_working_directory 

7from coxbuild.services import EventContext 

8 

9from . import delay, occur, periodic 

10 

11 

12class FileSystemChangeType(Enum): 

13 Create = 0 

14 Modify = 1 

15 Access = 2 

16 Delete = 3 

17 

18 

19@dataclass 

20class FileSystemEntry: 

21 path: Path 

22 creation: datetime 

23 modification: datetime 

24 access: datetime 

25 

26 def id(self): 

27 return str(self.path.resolve()) 

28 

29 

30@dataclass 

31class FileEntry(FileSystemEntry): 

32 pass 

33 

34 

35@dataclass 

36class DirectoryEntry(FileSystemEntry): 

37 pass 

38 

39 

40class FileSystemSnapshot: 

41 def __init__(self, path: Path, glob: str | None = None) -> None: 

42 self.entries: dict[str, FileSystemEntry] = {} 

43 items = path.glob(glob) if glob else [path] 

44 for item in items: 

45 item = item.resolve() 

46 stat = item.stat() 

47 ctime, mtime, atime = [datetime.fromtimestamp( 

48 t) for t in [stat.st_ctime, stat.st_mtime, stat.st_atime]] 

49 if item.is_file(): 

50 entry = FileEntry(item, ctime, mtime, atime) 

51 elif item.is_dir(): 

52 entry = DirectoryEntry(item, ctime, mtime, atime) 

53 if entry: 

54 self.entries[entry.id()] = entry 

55 

56 

57def diff(old: FileSystemSnapshot, new: FileSystemSnapshot): 

58 keys = set(old.entries) | set(new.entries) 

59 for key in keys: 

60 olde = old.entries.get(key) 

61 newe = new.entries.get(key) 

62 if not olde: 

63 yield (FileSystemChangeType.Create, newe) 

64 elif not newe: 

65 yield (FileSystemChangeType.Delete, olde) 

66 elif isinstance(olde, FileEntry) and isinstance(newe, FileEntry) or isinstance(olde, DirectoryEntry) and isinstance(newe, DirectoryEntry): 

67 if olde.creation < newe.creation: 

68 yield (FileSystemChangeType.Create, olde) 

69 if olde.modification < newe.modification: 

70 yield (FileSystemChangeType.Modify, olde) 

71 if olde.access < newe.access: 

72 yield (FileSystemChangeType.Access, olde) 

73 else: 

74 yield (FileSystemChangeType.Delete, olde) 

75 yield (FileSystemChangeType.Create, newe) 

76 

77 

78async def changed(path: Path | None = None, glob: str | None = None, type: set[FileSystemChangeType] | None = None, period: timedelta | None = None): 

79 """ 

80 Detect file or directory change (create, delete, modify, access). 

81 

82 Provide event context: 

83 type: Type of the change, in FileSystemChangeType 

84 entry: Changed entry 

85 

86 path: path to watch 

87 glob: glob pattern to watch 

88 type: change type to watch 

89 """ 

90 period = period or timedelta(seconds=0) 

91 path = path or get_working_directory() 

92 

93 snap = FileSystemSnapshot(path, glob) 

94 

95 async for _ in periodic(period): 

96 newsnap = FileSystemSnapshot(path, glob) 

97 

98 for ctype, entry in diff(snap, newsnap): 

99 if type is None or ctype in type: 

100 yield EventContext.build(type=ctype, entry=entry) 

101 

102 snap = newsnap 

103 

104 

105def access(path: Path | None = None, glob: str | None = None, period: timedelta | None = None): 

106 return changed(path, glob, {FileSystemChangeType.Access}, period) 

107 

108 

109def create(path: Path | None = None, glob: str | None = None, period: timedelta | None = None): 

110 return changed(path, glob, {FileSystemChangeType.Create}, period) 

111 

112 

113def modify(path: Path | None = None, glob: str | None = None, period: timedelta | None = None): 

114 return changed(path, glob, {FileSystemChangeType.Modify}, period) 

115 

116 

117def delete(path: Path | None = None, glob: str | None = None, period: timedelta | None = None): 

118 return changed(path, glob, {FileSystemChangeType.Delete}, period)