Coverage for src/lexigram/web/uploads/pipeline.py: 45%

1"""File upload handling and validation. 

2 

3Provides FileUpload model and validation pipeline. 

4""" 

5 

6from __future__ import annotations 

7 

8from abc import ABC, abstractmethod 

9from dataclasses import dataclass 

10from pathlib import Path 

11from typing import TYPE_CHECKING 

12 

13import aiofiles 

14import aiofiles.os 

15 

16from lexigram.result import Err, Ok, Result 

17 

18if TYPE_CHECKING: 

19 from starlette.datastructures import UploadFile 

20 

21 

22@dataclass(frozen=True) 

23class FileUpload: 

24 """Represents an uploaded file with metadata.""" 

25 

26 filename: str 

27 content_type: str 

28 size: int 

29 file: UploadFile 

30 

31 @property 

32 def extension(self) -> str: 

33 """Get file extension.""" 

34 return Path(self.filename).suffix.lower() 

35 

36 @property 

37 def name_without_extension(self) -> str: 

38 """Get filename without extension.""" 

39 return Path(self.filename).stem 

40 

41 async def read(self) -> bytes: 

42 """Read the entire file content.""" 

43 return await self.file.read() 

44 

45 async def save_to( 

46 self, path: str | Path, base_dir: Path | str | None = None 

47 ) -> None: 

48 """Save the file to a path. 

49 

50 Args: 

51 path: Destination path for the file. 

52 base_dir: Optional base directory. When provided, the resolved 

53 destination path must be relative to ``base_dir``; if it would 

54 escape that directory (e.g. via ``../../``), a 

55 :exc:`ValueError` is raised. 

56 

57 Raises: 

58 ValueError: If ``base_dir`` is provided and the resolved path 

59 escapes it (path-traversal prevention). 

60 """ 

61 path = Path(path) 

62 if base_dir is not None: 

63 resolved = path.resolve() 

64 base = Path(base_dir).resolve() 

65 if not resolved.is_relative_to(base): 

66 raise ValueError( 

67 f"Path {str(path)!r} escapes base directory {str(base_dir)!r}" 

68 ) 

69 await aiofiles.os.makedirs(path.parent, exist_ok=True) 

70 

71 content = await self.read() 

72 async with aiofiles.open(path, "wb") as f: 

73 await f.write(content) 

74 

75 async def stream_to( 

76 self, 

77 path: str | Path, 

78 chunk_size: int = 8192, 

79 base_dir: Path | str | None = None, 

80 ) -> None: 

81 """Stream file to disk in chunks. 

82 

83 Args: 

84 path: Destination path for the file. 

85 chunk_size: Number of bytes per read/write iteration. 

86 base_dir: Optional base directory. When provided, the resolved 

87 destination path must be relative to ``base_dir``; if it would 

88 escape that directory (e.g. via ``../../``), a 

89 :exc:`ValueError` is raised. 

90 

91 Raises: 

92 ValueError: If ``base_dir`` is provided and the resolved path 

93 escapes it (path-traversal prevention). 

94 """ 

95 path = Path(path) 

96 if base_dir is not None: 

97 resolved = path.resolve() 

98 base = Path(base_dir).resolve() 

99 if not resolved.is_relative_to(base): 

100 raise ValueError( 

101 f"Path {str(path)!r} escapes base directory {str(base_dir)!r}" 

102 ) 

103 await aiofiles.os.makedirs(path.parent, exist_ok=True) 

104 

105 async with aiofiles.open(path, "wb") as f: 

106 while True: 

107 chunk = await self.file.read(chunk_size) 

108 if not chunk: 

109 break 

110 await f.write(chunk) 

111 

112 

113class AbstractFileValidator(ABC): 

114 """Base class for file validation.""" 

115 

116 @abstractmethod 

117 async def validate(self, upload: FileUpload) -> Result[None, FileValidationError]: 

118 """Validate the file. 

119 

120 Args: 

121 upload: The file upload to validate. 

122 

123 Returns: 

124 Ok(None) if validation passes, Err(FileValidationError) if it fails. 

125 """ 

126 

127 

128@dataclass(frozen=True) 

129class FileValidationError: 

130 """Error type for file validation failures.""" 

131 

132 message: str 

133 

134 

135class FileSizeValidator(AbstractFileValidator): 

136 """Validates file size.""" 

137 

138 def __init__(self, max_size_bytes: int): 

139 self.max_size = max_size_bytes 

140 

141 async def validate(self, upload: FileUpload) -> Result[None, FileValidationError]: 

142 """Validate file size against the configured maximum. 

143 

144 Args: 

145 upload: The file upload to validate. 

146 

147 Returns: 

148 Ok(None) if size is within the limit, Err(FileValidationError) if exceeded. 

149 """ 

150 if upload.size > self.max_size: 

151 return Err( 

152 FileValidationError( 

153 message=f"File size {upload.size} exceeds maximum {self.max_size}", 

154 ) 

155 ) 

156 return Ok(None) 

157 

158 

159class FileTypeValidator(AbstractFileValidator): 

160 """Validates file MIME type.""" 

161 

162 def __init__(self, allowed_types: list[str]): 

163 self.allowed_types = [t.lower() for t in allowed_types] 

164 

165 async def validate(self, upload: FileUpload) -> Result[None, FileValidationError]: 

166 """Validate file MIME type against allowed types. 

167 

168 Args: 

169 upload: The file upload to validate. 

170 

171 Returns: 

172 Ok(None) if type is allowed, Err(FileValidationError) if disallowed. 

173 """ 

174 if upload.content_type.lower() not in self.allowed_types: 

175 return Err( 

176 FileValidationError( 

177 message=f"File type {upload.content_type} not allowed. " 

178 f"Allowed: {self.allowed_types}", 

179 ) 

180 ) 

181 return Ok(None) 

182 

183 

184class FileExtensionValidator(AbstractFileValidator): 

185 """Validates file extension.""" 

186 

187 def __init__(self, allowed_extensions: list[str]): 

188 self.allowed_extensions = [e.lower() for e in allowed_extensions] 

189 

190 async def validate(self, upload: FileUpload) -> Result[None, FileValidationError]: 

191 """Validate file extension against allowed extensions. 

192 

193 Args: 

194 upload: The file upload to validate. 

195 

196 Returns: 

197 Ok(None) if extension is allowed, Err(FileValidationError) if disallowed. 

198 """ 

199 if upload.extension not in self.allowed_extensions: 

200 return Err( 

201 FileValidationError( 

202 message=f"File extension {upload.extension} not allowed. " 

203 f"Allowed: {self.allowed_extensions}", 

204 ) 

205 ) 

206 return Ok(None) 

207 

208 

209class FileUploadPipeline: 

210 """Pipeline for processing file uploads with validation.""" 

211 

212 def __init__(self, validators: list[AbstractFileValidator] | None = None): 

213 self.validators = validators or [] 

214 

215 def add_validator(self, validator: AbstractFileValidator) -> FileUploadPipeline: 

216 """Add a validator to the pipeline.""" 

217 self.validators.append(validator) 

218 return self 

219 

220 async def process( 

221 self, upload: FileUpload 

222 ) -> Result[FileUpload, FileValidationError]: 

223 """Process and validate a file upload. 

224 

225 Args: 

226 upload: The file upload to process. 

227 

228 Returns: 

229 Ok(upload) if all validators pass, Err(FileValidationError) on first failure. 

230 """ 

231 for validator in self.validators: 

232 result = await validator.validate(upload) 

233 if result.is_err(): 

234 return Err(result.unwrap_err()) 

235 return Ok(upload) 

236 

237 async def process_multiple( 

238 self, 

239 uploads: list[FileUpload], 

240 ) -> list[Result[FileUpload, FileValidationError]]: 

241 """Process multiple file uploads.""" 

242 results = [] 

243 for upload in uploads: 

244 results.append(await self.process(upload)) 

245 return results