Mercurial > repos > shellac > guppy_basecaller
comparison env/lib/python3.7/site-packages/routes/util.py @ 5:9b1c78e6ba9c draft default tip
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author | shellac |
---|---|
date | Mon, 01 Jun 2020 08:59:25 -0400 |
parents | 79f47841a781 |
children |
comparison
equal
deleted
inserted
replaced
4:79f47841a781 | 5:9b1c78e6ba9c |
---|---|
1 """Utility functions for use in templates / controllers | |
2 | |
3 *PLEASE NOTE*: Many of these functions expect an initialized RequestConfig | |
4 object. This is expected to have been initialized for EACH REQUEST by the web | |
5 framework. | |
6 | |
7 """ | |
8 import os | |
9 import re | |
10 | |
11 import six | |
12 from six.moves import urllib | |
13 | |
14 from routes import request_config | |
15 | |
16 | |
17 class RoutesException(Exception): | |
18 """Tossed during Route exceptions""" | |
19 | |
20 | |
21 class MatchException(RoutesException): | |
22 """Tossed during URL matching exceptions""" | |
23 | |
24 | |
25 class GenerationException(RoutesException): | |
26 """Tossed during URL generation exceptions""" | |
27 | |
28 | |
29 def _screenargs(kargs, mapper, environ, force_explicit=False): | |
30 """ | |
31 Private function that takes a dict, and screens it against the current | |
32 request dict to determine what the dict should look like that is used. | |
33 This is responsible for the requests "memory" of the current. | |
34 """ | |
35 # Coerce any unicode args with the encoding | |
36 encoding = mapper.encoding | |
37 for key, val in six.iteritems(kargs): | |
38 if isinstance(val, six.text_type): | |
39 kargs[key] = val.encode(encoding) | |
40 | |
41 if mapper.explicit and mapper.sub_domains and not force_explicit: | |
42 return _subdomain_check(kargs, mapper, environ) | |
43 elif mapper.explicit and not force_explicit: | |
44 return kargs | |
45 | |
46 controller_name = as_unicode(kargs.get('controller'), encoding) | |
47 | |
48 if controller_name and controller_name.startswith('/'): | |
49 # If the controller name starts with '/', ignore route memory | |
50 kargs['controller'] = kargs['controller'][1:] | |
51 return kargs | |
52 elif controller_name and 'action' not in kargs: | |
53 # Fill in an action if we don't have one, but have a controller | |
54 kargs['action'] = 'index' | |
55 | |
56 route_args = environ.get('wsgiorg.routing_args') | |
57 if route_args: | |
58 memory_kargs = route_args[1].copy() | |
59 else: | |
60 memory_kargs = {} | |
61 | |
62 # Remove keys from memory and kargs if kargs has them as None | |
63 empty_keys = [key for key, value in six.iteritems(kargs) if value is None] | |
64 for key in empty_keys: | |
65 del kargs[key] | |
66 memory_kargs.pop(key, None) | |
67 | |
68 # Merge the new args on top of the memory args | |
69 memory_kargs.update(kargs) | |
70 | |
71 # Setup a sub-domain if applicable | |
72 if mapper.sub_domains: | |
73 memory_kargs = _subdomain_check(memory_kargs, mapper, environ) | |
74 return memory_kargs | |
75 | |
76 | |
77 def _subdomain_check(kargs, mapper, environ): | |
78 """Screen the kargs for a subdomain and alter it appropriately depending | |
79 on the current subdomain or lack therof.""" | |
80 if mapper.sub_domains: | |
81 subdomain = kargs.pop('sub_domain', None) | |
82 if isinstance(subdomain, six.text_type): | |
83 subdomain = str(subdomain) | |
84 | |
85 fullhost = environ.get('HTTP_HOST') or environ.get('SERVER_NAME') | |
86 | |
87 # In case environ defaulted to {} | |
88 if not fullhost: | |
89 return kargs | |
90 | |
91 hostmatch = fullhost.split(':') | |
92 host = hostmatch[0] | |
93 port = '' | |
94 if len(hostmatch) > 1: | |
95 port += ':' + hostmatch[1] | |
96 | |
97 match = re.match('^(.+?)\.(%s)$' % mapper.domain_match, host) | |
98 host_subdomain, domain = match.groups() if match else (None, host) | |
99 | |
100 subdomain = as_unicode(subdomain, mapper.encoding) | |
101 if subdomain and host_subdomain != subdomain and \ | |
102 subdomain not in mapper.sub_domains_ignore: | |
103 kargs['_host'] = subdomain + '.' + domain + port | |
104 elif (subdomain in mapper.sub_domains_ignore or \ | |
105 subdomain is None) and domain != host: | |
106 kargs['_host'] = domain + port | |
107 return kargs | |
108 else: | |
109 return kargs | |
110 | |
111 | |
112 def _url_quote(string, encoding): | |
113 """A Unicode handling version of urllib.quote.""" | |
114 if encoding: | |
115 if isinstance(string, six.text_type): | |
116 s = string.encode(encoding) | |
117 elif isinstance(string, six.text_type): | |
118 # assume the encoding is already correct | |
119 s = string | |
120 else: | |
121 s = six.text_type(string).encode(encoding) | |
122 else: | |
123 s = str(string) | |
124 return urllib.parse.quote(s, '/') | |
125 | |
126 | |
127 def _str_encode(string, encoding): | |
128 if encoding: | |
129 if isinstance(string, six.text_type): | |
130 s = string.encode(encoding) | |
131 elif isinstance(string, six.text_type): | |
132 # assume the encoding is already correct | |
133 s = string | |
134 else: | |
135 s = six.text_type(string).encode(encoding) | |
136 return s | |
137 | |
138 | |
139 def url_for(*args, **kargs): | |
140 """Generates a URL | |
141 | |
142 All keys given to url_for are sent to the Routes Mapper instance for | |
143 generation except for:: | |
144 | |
145 anchor specified the anchor name to be appened to the path | |
146 host overrides the default (current) host if provided | |
147 protocol overrides the default (current) protocol if provided | |
148 qualified creates the URL with the host/port information as | |
149 needed | |
150 | |
151 The URL is generated based on the rest of the keys. When generating a new | |
152 URL, values will be used from the current request's parameters (if | |
153 present). The following rules are used to determine when and how to keep | |
154 the current requests parameters: | |
155 | |
156 * If the controller is present and begins with '/', no defaults are used | |
157 * If the controller is changed, action is set to 'index' unless otherwise | |
158 specified | |
159 | |
160 For example, if the current request yielded a dict of | |
161 {'controller': 'blog', 'action': 'view', 'id': 2}, with the standard | |
162 ':controller/:action/:id' route, you'd get the following results:: | |
163 | |
164 url_for(id=4) => '/blog/view/4', | |
165 url_for(controller='/admin') => '/admin', | |
166 url_for(controller='admin') => '/admin/view/2' | |
167 url_for(action='edit') => '/blog/edit/2', | |
168 url_for(action='list', id=None) => '/blog/list' | |
169 | |
170 **Static and Named Routes** | |
171 | |
172 If there is a string present as the first argument, a lookup is done | |
173 against the named routes table to see if there's any matching routes. The | |
174 keyword defaults used with static routes will be sent in as GET query | |
175 arg's if a route matches. | |
176 | |
177 If no route by that name is found, the string is assumed to be a raw URL. | |
178 Should the raw URL begin with ``/`` then appropriate SCRIPT_NAME data will | |
179 be added if present, otherwise the string will be used as the url with | |
180 keyword args becoming GET query args. | |
181 | |
182 """ | |
183 anchor = kargs.get('anchor') | |
184 host = kargs.get('host') | |
185 protocol = kargs.pop('protocol', None) | |
186 qualified = kargs.pop('qualified', None) | |
187 | |
188 # Remove special words from kargs, convert placeholders | |
189 for key in ['anchor', 'host']: | |
190 if kargs.get(key): | |
191 del kargs[key] | |
192 if key+'_' in kargs: | |
193 kargs[key] = kargs.pop(key+'_') | |
194 | |
195 if 'protocol_' in kargs: | |
196 kargs['protocol_'] = protocol | |
197 | |
198 config = request_config() | |
199 route = None | |
200 static = False | |
201 encoding = config.mapper.encoding | |
202 url = '' | |
203 if len(args) > 0: | |
204 route = config.mapper._routenames.get(args[0]) | |
205 | |
206 # No named route found, assume the argument is a relative path | |
207 if not route: | |
208 static = True | |
209 url = args[0] | |
210 | |
211 if url.startswith('/') and hasattr(config, 'environ') \ | |
212 and config.environ.get('SCRIPT_NAME'): | |
213 url = config.environ.get('SCRIPT_NAME') + url | |
214 | |
215 if static: | |
216 if kargs: | |
217 url += '?' | |
218 query_args = [] | |
219 for key, val in six.iteritems(kargs): | |
220 if isinstance(val, (list, tuple)): | |
221 for value in val: | |
222 query_args.append("%s=%s" % ( | |
223 urllib.parse.quote(six.text_type(key).encode(encoding)), | |
224 urllib.parse.quote(six.text_type(value).encode(encoding)))) | |
225 else: | |
226 query_args.append("%s=%s" % ( | |
227 urllib.parse.quote(six.text_type(key).encode(encoding)), | |
228 urllib.parse.quote(six.text_type(val).encode(encoding)))) | |
229 url += '&'.join(query_args) | |
230 environ = getattr(config, 'environ', {}) | |
231 if 'wsgiorg.routing_args' not in environ: | |
232 environ = environ.copy() | |
233 mapper_dict = getattr(config, 'mapper_dict', None) | |
234 if mapper_dict is not None: | |
235 match_dict = mapper_dict.copy() | |
236 else: | |
237 match_dict = {} | |
238 environ['wsgiorg.routing_args'] = ((), match_dict) | |
239 | |
240 if not static: | |
241 route_args = [] | |
242 if route: | |
243 if config.mapper.hardcode_names: | |
244 route_args.append(route) | |
245 newargs = route.defaults.copy() | |
246 newargs.update(kargs) | |
247 | |
248 # If this route has a filter, apply it | |
249 if route.filter: | |
250 newargs = route.filter(newargs) | |
251 | |
252 if not route.static: | |
253 # Handle sub-domains | |
254 newargs = _subdomain_check(newargs, config.mapper, environ) | |
255 else: | |
256 newargs = _screenargs(kargs, config.mapper, environ) | |
257 anchor = newargs.pop('_anchor', None) or anchor | |
258 host = newargs.pop('_host', None) or host | |
259 protocol = newargs.pop('_protocol', protocol) | |
260 url = config.mapper.generate(*route_args, **newargs) | |
261 if anchor is not None: | |
262 url += '#' + _url_quote(anchor, encoding) | |
263 if host or (protocol is not None) or qualified: | |
264 if not host and not qualified: | |
265 # Ensure we don't use a specific port, as changing the protocol | |
266 # means that we most likely need a new port | |
267 host = config.host.split(':')[0] | |
268 elif not host: | |
269 host = config.host | |
270 if protocol is None: | |
271 protocol = config.protocol | |
272 if protocol != '': | |
273 protocol += ':' | |
274 if url is not None: | |
275 url = protocol + '//' + host + url | |
276 | |
277 if not ascii_characters(url) and url is not None: | |
278 raise GenerationException("url_for can only return a string, got " | |
279 "unicode instead: %s" % url) | |
280 if url is None: | |
281 raise GenerationException( | |
282 "url_for could not generate URL. Called with args: %s %s" % \ | |
283 (args, kargs)) | |
284 return url | |
285 | |
286 | |
287 class URLGenerator(object): | |
288 """The URL Generator generates URL's | |
289 | |
290 It is automatically instantiated by the RoutesMiddleware and put | |
291 into the ``wsgiorg.routing_args`` tuple accessible as:: | |
292 | |
293 url = environ['wsgiorg.routing_args'][0][0] | |
294 | |
295 Or via the ``routes.url`` key:: | |
296 | |
297 url = environ['routes.url'] | |
298 | |
299 The url object may be instantiated outside of a web context for use | |
300 in testing, however sub_domain support and fully qualified URL's | |
301 cannot be generated without supplying a dict that must contain the | |
302 key ``HTTP_HOST``. | |
303 | |
304 """ | |
305 def __init__(self, mapper, environ): | |
306 """Instantiate the URLGenerator | |
307 | |
308 ``mapper`` | |
309 The mapper object to use when generating routes. | |
310 ``environ`` | |
311 The environment dict used in WSGI, alternately, any dict | |
312 that contains at least an ``HTTP_HOST`` value. | |
313 | |
314 """ | |
315 self.mapper = mapper | |
316 if 'SCRIPT_NAME' not in environ: | |
317 environ['SCRIPT_NAME'] = '' | |
318 self.environ = environ | |
319 | |
320 def __call__(self, *args, **kargs): | |
321 """Generates a URL | |
322 | |
323 All keys given to url_for are sent to the Routes Mapper instance for | |
324 generation except for:: | |
325 | |
326 anchor specified the anchor name to be appened to the path | |
327 host overrides the default (current) host if provided | |
328 protocol overrides the default (current) protocol if provided | |
329 qualified creates the URL with the host/port information as | |
330 needed | |
331 | |
332 """ | |
333 anchor = kargs.get('anchor') | |
334 host = kargs.get('host') | |
335 protocol = kargs.pop('protocol', None) | |
336 qualified = kargs.pop('qualified', None) | |
337 | |
338 # Remove special words from kargs, convert placeholders | |
339 for key in ['anchor', 'host']: | |
340 if kargs.get(key): | |
341 del kargs[key] | |
342 if key+'_' in kargs: | |
343 kargs[key] = kargs.pop(key+'_') | |
344 | |
345 if 'protocol_' in kargs: | |
346 kargs['protocol_'] = protocol | |
347 | |
348 route = None | |
349 use_current = '_use_current' in kargs and kargs.pop('_use_current') | |
350 | |
351 static = False | |
352 encoding = self.mapper.encoding | |
353 url = '' | |
354 | |
355 more_args = len(args) > 0 | |
356 if more_args: | |
357 route = self.mapper._routenames.get(args[0]) | |
358 | |
359 if not route and more_args: | |
360 static = True | |
361 url = args[0] | |
362 if url.startswith('/') and self.environ.get('SCRIPT_NAME'): | |
363 url = self.environ.get('SCRIPT_NAME') + url | |
364 | |
365 if static: | |
366 if kargs: | |
367 url += '?' | |
368 query_args = [] | |
369 for key, val in six.iteritems(kargs): | |
370 if isinstance(val, (list, tuple)): | |
371 for value in val: | |
372 query_args.append("%s=%s" % ( | |
373 urllib.parse.quote(six.text_type(key).encode(encoding)), | |
374 urllib.parse.quote(six.text_type(value).encode(encoding)))) | |
375 else: | |
376 query_args.append("%s=%s" % ( | |
377 urllib.parse.quote(six.text_type(key).encode(encoding)), | |
378 urllib.parse.quote(six.text_type(val).encode(encoding)))) | |
379 url += '&'.join(query_args) | |
380 if not static: | |
381 route_args = [] | |
382 if route: | |
383 if self.mapper.hardcode_names: | |
384 route_args.append(route) | |
385 newargs = route.defaults.copy() | |
386 newargs.update(kargs) | |
387 | |
388 # If this route has a filter, apply it | |
389 if route.filter: | |
390 newargs = route.filter(newargs) | |
391 if not route.static or (route.static and not route.external): | |
392 # Handle sub-domains, retain sub_domain if there is one | |
393 sub = newargs.get('sub_domain', None) | |
394 newargs = _subdomain_check(newargs, self.mapper, | |
395 self.environ) | |
396 # If the route requires a sub-domain, and we have it, restore | |
397 # it | |
398 if 'sub_domain' in route.defaults: | |
399 newargs['sub_domain'] = sub | |
400 | |
401 elif use_current: | |
402 newargs = _screenargs(kargs, self.mapper, self.environ, force_explicit=True) | |
403 elif 'sub_domain' in kargs: | |
404 newargs = _subdomain_check(kargs, self.mapper, self.environ) | |
405 else: | |
406 newargs = kargs | |
407 | |
408 anchor = anchor or newargs.pop('_anchor', None) | |
409 host = host or newargs.pop('_host', None) | |
410 if protocol is None: | |
411 protocol = newargs.pop('_protocol', None) | |
412 newargs['_environ'] = self.environ | |
413 url = self.mapper.generate(*route_args, **newargs) | |
414 if anchor is not None: | |
415 url += '#' + _url_quote(anchor, encoding) | |
416 if host or (protocol is not None) or qualified: | |
417 if 'routes.cached_hostinfo' not in self.environ: | |
418 cache_hostinfo(self.environ) | |
419 hostinfo = self.environ['routes.cached_hostinfo'] | |
420 | |
421 if not host and not qualified: | |
422 # Ensure we don't use a specific port, as changing the protocol | |
423 # means that we most likely need a new port | |
424 host = hostinfo['host'].split(':')[0] | |
425 elif not host: | |
426 host = hostinfo['host'] | |
427 if protocol is None: | |
428 protocol = hostinfo['protocol'] | |
429 if protocol != '': | |
430 protocol += ':' | |
431 if url is not None: | |
432 if host[-1] != '/': | |
433 host += '/' | |
434 url = protocol + '//' + host + url.lstrip('/') | |
435 | |
436 if not ascii_characters(url) and url is not None: | |
437 raise GenerationException("Can only return a string, got " | |
438 "unicode instead: %s" % url) | |
439 if url is None: | |
440 raise GenerationException( | |
441 "Could not generate URL. Called with args: %s %s" % \ | |
442 (args, kargs)) | |
443 return url | |
444 | |
445 def current(self, *args, **kwargs): | |
446 """Generate a route that includes params used on the current | |
447 request | |
448 | |
449 The arguments for this method are identical to ``__call__`` | |
450 except that arguments set to None will remove existing route | |
451 matches of the same name from the set of arguments used to | |
452 construct a URL. | |
453 """ | |
454 return self(_use_current=True, *args, **kwargs) | |
455 | |
456 | |
457 def redirect_to(*args, **kargs): | |
458 """Issues a redirect based on the arguments. | |
459 | |
460 Redirect's *should* occur as a "302 Moved" header, however the web | |
461 framework may utilize a different method. | |
462 | |
463 All arguments are passed to url_for to retrieve the appropriate URL, then | |
464 the resulting URL it sent to the redirect function as the URL. | |
465 """ | |
466 target = url_for(*args, **kargs) | |
467 config = request_config() | |
468 return config.redirect(target) | |
469 | |
470 | |
471 def cache_hostinfo(environ): | |
472 """Processes the host information and stores a copy | |
473 | |
474 This work was previously done but wasn't stored in environ, nor is | |
475 it guaranteed to be setup in the future (Routes 2 and beyond). | |
476 | |
477 cache_hostinfo processes environ keys that may be present to | |
478 determine the proper host, protocol, and port information to use | |
479 when generating routes. | |
480 | |
481 """ | |
482 hostinfo = {} | |
483 if environ.get('HTTPS') or environ.get('wsgi.url_scheme') == 'https' \ | |
484 or 'https' in environ.get('HTTP_X_FORWARDED_PROTO', "").split(', '): | |
485 hostinfo['protocol'] = 'https' | |
486 else: | |
487 hostinfo['protocol'] = 'http' | |
488 if environ.get('HTTP_X_FORWARDED_HOST'): | |
489 hostinfo['host'] = environ['HTTP_X_FORWARDED_HOST'].split(', ', 1)[0] | |
490 elif environ.get('HTTP_HOST'): | |
491 hostinfo['host'] = environ['HTTP_HOST'] | |
492 else: | |
493 hostinfo['host'] = environ['SERVER_NAME'] | |
494 if environ.get('wsgi.url_scheme') == 'https': | |
495 if environ['SERVER_PORT'] != '443': | |
496 hostinfo['host'] += ':' + environ['SERVER_PORT'] | |
497 else: | |
498 if environ['SERVER_PORT'] != '80': | |
499 hostinfo['host'] += ':' + environ['SERVER_PORT'] | |
500 environ['routes.cached_hostinfo'] = hostinfo | |
501 return hostinfo | |
502 | |
503 | |
504 def controller_scan(directory=None): | |
505 """Scan a directory for python files and use them as controllers""" | |
506 if directory is None: | |
507 return [] | |
508 | |
509 def find_controllers(dirname, prefix=''): | |
510 """Locate controllers in a directory""" | |
511 controllers = [] | |
512 for fname in os.listdir(dirname): | |
513 filename = os.path.join(dirname, fname) | |
514 if os.path.isfile(filename) and \ | |
515 re.match('^[^_]{1,1}.*\.py$', fname): | |
516 controllers.append(prefix + fname[:-3]) | |
517 elif os.path.isdir(filename): | |
518 controllers.extend(find_controllers(filename, | |
519 prefix=prefix+fname+'/')) | |
520 return controllers | |
521 controllers = find_controllers(directory) | |
522 # Sort by string length, shortest goes first | |
523 controllers.sort(key=len, reverse=True) | |
524 return controllers | |
525 | |
526 | |
527 def as_unicode(value, encoding, errors='strict'): | |
528 if value is not None and isinstance(value, bytes): | |
529 return value.decode(encoding, errors) | |
530 | |
531 return value | |
532 | |
533 | |
534 def ascii_characters(string): | |
535 if string is None: | |
536 return True | |
537 | |
538 return all(ord(c) < 128 for c in string) |