Coverage for src/lexigram/web/pipes/builtin/file.py: 30%

1"""File validation pipes for file uploads. 

2 

3Pipes that validate file uploads (size, type, etc). 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any 

9 

10from lexigram.web.protocols import ParamMetadata, PipeProtocol 

11 

12 

13class FileSizeValidationPipe(PipeProtocol): 

14 """PipeProtocol that validates file upload size. 

15 

16 Example: 

17 ```python 

18 class UploadController(Controller): 

19 @post("/upload") 

20 async def upload( 

21 self, 

22 @file(pipe=FileSizeValidationPipe(max_size_mb=10)) file: UploadFile 

23 ): 

24 ... 

25 ``` 

26 """ 

27 

28 def __init__(self, max_size_mb: int = 10): 

29 """Initialize the pipe. 

30 

31 Args: 

32 max_size_mb: Maximum file size in megabytes. 

33 """ 

34 self._max_size = max_size_mb * 1024 * 1024 # Convert to bytes 

35 

36 async def transform(self, value: Any, metadata: ParamMetadata) -> Any: 

37 """Validate file size. 

38 

39 Args: 

40 value: The uploaded file. 

41 metadata: Metadata about the parameter. 

42 

43 Returns: 

44 The file if valid. 

45 

46 Raises: 

47 ValueError: If file exceeds max size. 

48 """ 

49 if value is None: 

50 return None 

51 

52 # Check if it's a file-like object 

53 if hasattr(value, "size"): 

54 if value.size > self._max_size: 

55 raise ValueError( 

56 f"File '{getattr(value, 'filename', 'unknown')}' " 

57 f"exceeds maximum size of {self._max_size // (1024 * 1024)}MB", 

58 ) 

59 elif hasattr(value, "content_length"): 

60 if value.content_length and value.content_length > self._max_size: 

61 raise ValueError( 

62 f"File exceeds maximum size of {self._max_size // (1024 * 1024)}MB", 

63 ) 

64 

65 return value 

66 

67 

68class FileTypeValidationPipe(PipeProtocol): 

69 """PipeProtocol that validates file MIME types. 

70 

71 Example: 

72 ```python 

73 class UploadController(Controller): 

74 @post("/upload") 

75 async def upload( 

76 self, 

77 @file(pipe=FileTypeValidationPipe(allowed_types=["image/png", "image/jpeg"])) file: UploadFile 

78 ): 

79 ... 

80 ``` 

81 """ 

82 

83 def __init__(self, allowed_types: list[str] | None = None): 

84 """Initialize the pipe. 

85 

86 Args: 

87 allowed_types: List of allowed MIME types. If None, accepts all. 

88 """ 

89 self._allowed_types = allowed_types or [] 

90 

91 async def transform(self, value: Any, metadata: ParamMetadata) -> Any: 

92 """Validate file type. 

93 

94 Args: 

95 value: The uploaded file. 

96 metadata: Metadata about the parameter. 

97 

98 Returns: 

99 The file if valid. 

100 

101 Raises: 

102 ValueError: If file type is not allowed. 

103 """ 

104 if value is None: 

105 return None 

106 

107 if not self._allowed_types: 

108 return value 

109 

110 # Get content type 

111 content_type = None 

112 if hasattr(value, "content_type"): 

113 content_type = value.content_type 

114 elif hasattr(value, "headers"): 

115 content_type = value.headers.get("content-type") 

116 

117 if content_type and content_type not in self._allowed_types: 

118 raise ValueError( 

119 f"File type '{content_type}' not allowed. " 

120 f"Allowed types: {', '.join(self._allowed_types)}", 

121 ) 

122 

123 return value 

124 

125 

126class TrimPipe(PipeProtocol): 

127 """PipeProtocol that trims whitespace from string inputs. 

128 

129 Example: 

130 ```python 

131 class SearchController(Controller): 

132 @get("/search") 

133 async def search( 

134 self, 

135 @query(pipe=TrimPipe()) query: str, 

136 ): 

137 ... 

138 ``` 

139 """ 

140 

141 async def transform(self, value: Any, metadata: ParamMetadata) -> Any: 

142 """Trim whitespace from string. 

143 

144 Args: 

145 value: The value to trim. 

146 metadata: Metadata about the parameter. 

147 

148 Returns: 

149 Trimmed string. 

150 """ 

151 if value is None: 

152 return None 

153 

154 if isinstance(value, str): 

155 return value.strip() 

156 

157 return value 

158 

159 

160__all__ = [ 

161 "FileSizeValidationPipe", 

162 "FileTypeValidationPipe", 

163 "TrimPipe", 

164]