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 getRequestBody(self) -> str:
44	def getRequestBody(self) -> str:
45		return self.__requestBody
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.

def getFile(self, name: str) -> Wsgi_15Watt.multipart.MultipartPart:
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.

def getDíctFiles(self) -> Dict[str, Wsgi_15Watt.multipart.MultipartPart]:
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

def getEnvByKey(self, key: str) -> str | None:
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)

Returns the value of key from the env-Dict or None.

env: dict
148	@property
149	def env(self) -> dict:
150		"""
151			Returns the environment dictionary.
152		"""
153		return self.__env

Returns the environment dictionary.