Wsgi_15Watt.Request
1from typing import Dict 2from urllib.parse import unquote, parse_qs 3from io import BytesIO 4from .multipart import parse_form_data, MultipartPart 5from .Exceptions import ParamNotFound, ValueNotFound, FileNotFound 6 7 8class Request(object): 9 """ 10 Represents the http request. 11 """ 12 def __init__(self, env: dict, paramsFromRoute: dict): 13 self.__env = env 14 self.__header = {} 15 self.__user = self.__user = env.get('REMOTE_USER', '') 16 self.__password = '' 17 18 self.__params, self.__files = self.__getParameters(env, paramsFromRoute) 19 20 self.__requestBodySize = 0 21 self.__requestBody = '' 22 23 try: 24 self.__requestBodySize = int(self.__env.get('CONTENT_LENGTH', 0)) 25 except ValueError: 26 pass 27 except TypeError: 28 pass 29 30 byteIo = BytesIO(initial_bytes=env['Wsgi_15Watt.input'].read(self.__requestBodySize)) 31 32 # Read the requests body no matter what content type or method the request is 33 byteIo.seek(0) 34 self.__requestBody = unquote(byteIo.read(self.__requestBodySize).decode('utf-8')) 35 byteIo.close() 36 37 # Methode 38 self.__requestMethod = self.__env.get('REQUEST_METHOD', 'GET') 39 40 return 41 42 43 def getRequestBody(self) -> str: 44 return self.__requestBody 45 46 47 def get(self, name: str): 48 """ 49 Returns the value of the parameter name. 50 If there are multiple values with the same name, the first value is returned. 51 52 :raise: ParamNotFound 53 """ 54 if name not in self.__params: 55 raise ParamNotFound(returnMsg=f'Parameter {name} not found') 56 57 if 0 == len(self.__params[name]): 58 raise ParamNotFound(returnMsg=f'Parameter {name} is empty') 59 60 return self.__params[name][0] 61 62 63 def getAsList(self, name: str) -> list: 64 """ 65 Returns the list of values of the parameter name. 66 67 :raise: ParamNotFound 68 """ 69 if name not in self.__params: 70 raise ParamNotFound(returnMsg=f'Parameter {name} not found') 71 72 if 0 == len(self.__params[name]): 73 raise ParamNotFound(returnMsg=f'Parameter {name} is empty') 74 75 return self.__params[name] 76 77 78 def getDictParams(self) -> dict: 79 """ 80 Returns all parameters as a dictionary. 81 """ 82 return self.__params 83 84 85 def has(self, name: str) -> bool: 86 """ 87 Checks if the parameter name exists. 88 """ 89 return name in self.__params 90 91 92 def hasFile(self, name: str) -> bool: 93 """ 94 Checks if a file with the name exists. 95 """ 96 return name in self.__files 97 98 99 def getFile(self, name: str) -> MultipartPart: 100 """ 101 Returns the file with the name. 102 """ 103 if name not in self.__files: 104 raise FileNotFound(returnMsg=f'File {name} not found') 105 106 return self.__files[name] 107 108 109 def getDíctFiles(self) -> Dict[str, MultipartPart]: 110 """ 111 Returns all files as a dictionary. 112 """ 113 return self.__files 114 115 116 def getHeader(self, name: str) -> str: 117 """ 118 Returns the value of the header field name. 119 """ 120 if name not in self.__header: 121 raise ValueNotFound(returnMsg=f'Header-Field {name} not found') 122 123 return self.__header[name] 124 125 126 def hasHeader(self, name: str) -> bool: 127 """ 128 Check if there is a value for name in the header values. 129 """ 130 return name in self.__header 131 132 133 def envHasKey(self, key: str) -> bool: 134 """ 135 Check if the key exists in the env-Dict 136 """ 137 return key in self.__env 138 139 140 def getEnvByKey(self, key: str) -> str|None: 141 """ 142 Returns the value of key from the env-Dict or None. 143 """ 144 return self.__env.get(key, None) 145 146 147 @property 148 def env(self) -> dict: 149 """ 150 Returns the environment dictionary. 151 """ 152 return self.__env 153 154 155 def __getParameters(self, env: dict, paramsFromRoute: dict) -> (dict, dict): 156 """ 157 Reads the parameters from all sources: 158 - from the query string (GET) 159 - from the request body (POST, PUT) 160 - from the path (Route) 161 162 Each parameter can have multiple values. 163 Files will be stored as a parameter und in files 164 """ 165 dictParams = {} 166 dictFiles = {} 167 168 # Add the parameters from the route 169 for key in paramsFromRoute: 170 dictParams[key] = [paramsFromRoute[key]] 171 172 # Add the parameters from the query string 173 queryParams = parse_qs(env.get('QUERY_STRING', '')) 174 for item in queryParams.items(): 175 if item[0] in dictParams: 176 dictParams[item[0]] += item[1] 177 else: 178 dictParams[item[0]] = item[1] 179 180 # Add the parameters from the request body 181 (dictForm, dictMultiFiles) = parse_form_data(environ=env) 182 183 for key in dictForm: 184 if key in dictParams: 185 dictParams[key] += dictForm.getall(key) 186 else: 187 dictParams[key] = dictForm.getall(key) 188 189 for key in dictMultiFiles: 190 if key in dictParams: 191 dictParams[key] += [dictMultiFiles[key]] 192 else: 193 dictParams[key] = [dictMultiFiles[key]] 194 195 # Save files also separately 196 dictFiles[key] = dictMultiFiles[key] 197 198 return dictParams, dictFiles
class
Request:
9class Request(object): 10 """ 11 Represents the http request. 12 """ 13 def __init__(self, env: dict, paramsFromRoute: dict): 14 self.__env = env 15 self.__header = {} 16 self.__user = self.__user = env.get('REMOTE_USER', '') 17 self.__password = '' 18 19 self.__params, self.__files = self.__getParameters(env, paramsFromRoute) 20 21 self.__requestBodySize = 0 22 self.__requestBody = '' 23 24 try: 25 self.__requestBodySize = int(self.__env.get('CONTENT_LENGTH', 0)) 26 except ValueError: 27 pass 28 except TypeError: 29 pass 30 31 byteIo = BytesIO(initial_bytes=env['Wsgi_15Watt.input'].read(self.__requestBodySize)) 32 33 # Read the requests body no matter what content type or method the request is 34 byteIo.seek(0) 35 self.__requestBody = unquote(byteIo.read(self.__requestBodySize).decode('utf-8')) 36 byteIo.close() 37 38 # Methode 39 self.__requestMethod = self.__env.get('REQUEST_METHOD', 'GET') 40 41 return 42 43 44 def getRequestBody(self) -> str: 45 return self.__requestBody 46 47 48 def get(self, name: str): 49 """ 50 Returns the value of the parameter name. 51 If there are multiple values with the same name, the first value is returned. 52 53 :raise: ParamNotFound 54 """ 55 if name not in self.__params: 56 raise ParamNotFound(returnMsg=f'Parameter {name} not found') 57 58 if 0 == len(self.__params[name]): 59 raise ParamNotFound(returnMsg=f'Parameter {name} is empty') 60 61 return self.__params[name][0] 62 63 64 def getAsList(self, name: str) -> list: 65 """ 66 Returns the list of values of the parameter name. 67 68 :raise: ParamNotFound 69 """ 70 if name not in self.__params: 71 raise ParamNotFound(returnMsg=f'Parameter {name} not found') 72 73 if 0 == len(self.__params[name]): 74 raise ParamNotFound(returnMsg=f'Parameter {name} is empty') 75 76 return self.__params[name] 77 78 79 def getDictParams(self) -> dict: 80 """ 81 Returns all parameters as a dictionary. 82 """ 83 return self.__params 84 85 86 def has(self, name: str) -> bool: 87 """ 88 Checks if the parameter name exists. 89 """ 90 return name in self.__params 91 92 93 def hasFile(self, name: str) -> bool: 94 """ 95 Checks if a file with the name exists. 96 """ 97 return name in self.__files 98 99 100 def getFile(self, name: str) -> MultipartPart: 101 """ 102 Returns the file with the name. 103 """ 104 if name not in self.__files: 105 raise FileNotFound(returnMsg=f'File {name} not found') 106 107 return self.__files[name] 108 109 110 def getDíctFiles(self) -> Dict[str, MultipartPart]: 111 """ 112 Returns all files as a dictionary. 113 """ 114 return self.__files 115 116 117 def getHeader(self, name: str) -> str: 118 """ 119 Returns the value of the header field name. 120 """ 121 if name not in self.__header: 122 raise ValueNotFound(returnMsg=f'Header-Field {name} not found') 123 124 return self.__header[name] 125 126 127 def hasHeader(self, name: str) -> bool: 128 """ 129 Check if there is a value for name in the header values. 130 """ 131 return name in self.__header 132 133 134 def envHasKey(self, key: str) -> bool: 135 """ 136 Check if the key exists in the env-Dict 137 """ 138 return key in self.__env 139 140 141 def getEnvByKey(self, key: str) -> str|None: 142 """ 143 Returns the value of key from the env-Dict or None. 144 """ 145 return self.__env.get(key, None) 146 147 148 @property 149 def env(self) -> dict: 150 """ 151 Returns the environment dictionary. 152 """ 153 return self.__env 154 155 156 def __getParameters(self, env: dict, paramsFromRoute: dict) -> (dict, dict): 157 """ 158 Reads the parameters from all sources: 159 - from the query string (GET) 160 - from the request body (POST, PUT) 161 - from the path (Route) 162 163 Each parameter can have multiple values. 164 Files will be stored as a parameter und in files 165 """ 166 dictParams = {} 167 dictFiles = {} 168 169 # Add the parameters from the route 170 for key in paramsFromRoute: 171 dictParams[key] = [paramsFromRoute[key]] 172 173 # Add the parameters from the query string 174 queryParams = parse_qs(env.get('QUERY_STRING', '')) 175 for item in queryParams.items(): 176 if item[0] in dictParams: 177 dictParams[item[0]] += item[1] 178 else: 179 dictParams[item[0]] = item[1] 180 181 # Add the parameters from the request body 182 (dictForm, dictMultiFiles) = parse_form_data(environ=env) 183 184 for key in dictForm: 185 if key in dictParams: 186 dictParams[key] += dictForm.getall(key) 187 else: 188 dictParams[key] = dictForm.getall(key) 189 190 for key in dictMultiFiles: 191 if key in dictParams: 192 dictParams[key] += [dictMultiFiles[key]] 193 else: 194 dictParams[key] = [dictMultiFiles[key]] 195 196 # Save files also separately 197 dictFiles[key] = dictMultiFiles[key] 198 199 return dictParams, dictFiles
Represents the http request.
Request(env: dict, paramsFromRoute: dict)
13 def __init__(self, env: dict, paramsFromRoute: dict): 14 self.__env = env 15 self.__header = {} 16 self.__user = self.__user = env.get('REMOTE_USER', '') 17 self.__password = '' 18 19 self.__params, self.__files = self.__getParameters(env, paramsFromRoute) 20 21 self.__requestBodySize = 0 22 self.__requestBody = '' 23 24 try: 25 self.__requestBodySize = int(self.__env.get('CONTENT_LENGTH', 0)) 26 except ValueError: 27 pass 28 except TypeError: 29 pass 30 31 byteIo = BytesIO(initial_bytes=env['Wsgi_15Watt.input'].read(self.__requestBodySize)) 32 33 # Read the requests body no matter what content type or method the request is 34 byteIo.seek(0) 35 self.__requestBody = unquote(byteIo.read(self.__requestBodySize).decode('utf-8')) 36 byteIo.close() 37 38 # Methode 39 self.__requestMethod = self.__env.get('REQUEST_METHOD', 'GET') 40 41 return
def
get(self, name: str):
48 def get(self, name: str): 49 """ 50 Returns the value of the parameter name. 51 If there are multiple values with the same name, the first value is returned. 52 53 :raise: ParamNotFound 54 """ 55 if name not in self.__params: 56 raise ParamNotFound(returnMsg=f'Parameter {name} not found') 57 58 if 0 == len(self.__params[name]): 59 raise ParamNotFound(returnMsg=f'Parameter {name} is empty') 60 61 return self.__params[name][0]
Returns the value of the parameter name. If there are multiple values with the same name, the first value is returned.
:raise: ParamNotFound
def
getAsList(self, name: str) -> list:
64 def getAsList(self, name: str) -> list: 65 """ 66 Returns the list of values of the parameter name. 67 68 :raise: ParamNotFound 69 """ 70 if name not in self.__params: 71 raise ParamNotFound(returnMsg=f'Parameter {name} not found') 72 73 if 0 == len(self.__params[name]): 74 raise ParamNotFound(returnMsg=f'Parameter {name} is empty') 75 76 return self.__params[name]
Returns the list of values of the parameter name.
:raise: ParamNotFound
def
getDictParams(self) -> dict:
79 def getDictParams(self) -> dict: 80 """ 81 Returns all parameters as a dictionary. 82 """ 83 return self.__params
Returns all parameters as a dictionary.
def
has(self, name: str) -> bool:
86 def has(self, name: str) -> bool: 87 """ 88 Checks if the parameter name exists. 89 """ 90 return name in self.__params
Checks if the parameter name exists.
def
hasFile(self, name: str) -> bool:
93 def hasFile(self, name: str) -> bool: 94 """ 95 Checks if a file with the name exists. 96 """ 97 return name in self.__files
Checks if a file with the name exists.
100 def getFile(self, name: str) -> MultipartPart: 101 """ 102 Returns the file with the name. 103 """ 104 if name not in self.__files: 105 raise FileNotFound(returnMsg=f'File {name} not found') 106 107 return self.__files[name]
Returns the file with the name.
110 def getDíctFiles(self) -> Dict[str, MultipartPart]: 111 """ 112 Returns all files as a dictionary. 113 """ 114 return self.__files
Returns all files as a dictionary.
def
getHeader(self, name: str) -> str:
117 def getHeader(self, name: str) -> str: 118 """ 119 Returns the value of the header field name. 120 """ 121 if name not in self.__header: 122 raise ValueNotFound(returnMsg=f'Header-Field {name} not found') 123 124 return self.__header[name]
Returns the value of the header field name.
def
hasHeader(self, name: str) -> bool:
127 def hasHeader(self, name: str) -> bool: 128 """ 129 Check if there is a value for name in the header values. 130 """ 131 return name in self.__header
Check if there is a value for name in the header values.
def
envHasKey(self, key: str) -> bool:
134 def envHasKey(self, key: str) -> bool: 135 """ 136 Check if the key exists in the env-Dict 137 """ 138 return key in self.__env
Check if the key exists in the env-Dict