Python re 模块，MatchObject() 实例源码
我们从Python开源项目中，提取了以下16个代码示例，用于说明如何使用re.MatchObject()。
def ExtractVariableFromMatch(re_match_obj):
"""Takes a match from VARIABLE_REFERENCE_RE and extracts the variable.
This funciton is exposed to testing.
Args:
re_match_obj: a re.MatchObject
Returns:
The variable name (with qualifier attached) or None if the match wasn't
completely valid.
"""
expected_close = '}' if re_match_obj.group(1) == '{' else ')'
if re_match_obj.group(3) == expected_close:
m = VARIABLE_NAME_RE.match(re_match_obj.group(2))
if m:
return m.group(0)
return None
def regexMatch(render, pattern, string, flags = 0):
"""
Jinja2 global: Match a string for regular expression pattern.
This function internally runs re.match().
:param s: String where to be searched in.
:type s: str
:param pattern: Regular expression pattern to be matched.
:type pattern: str
:param flags: Flags to be passed to re.search().
:type flags: int
:return: A matching object on success, else None.
:rtype: ``re.MatchObject``
"""
return re.match(pattern,string)
def regexSearch(render, s, pattern, flags = 0):
"""
Jinja2 global: Search a string for regular expression pattern.
This function internally runs re.search().
:param s: String where to be searched in.
:type s: str
:param pattern: Regular expression pattern to be matched.
:type pattern: str
:param flags: Flags to be passed to re.search().
:type flags: int
:return: A matching object on success, else None.
:rtype: ``re.MatchObject``
"""
return re.search(pattern, s, flags)
def _handle_script_request(self,
environ,
start_response,
url_map,
match,
inst=None):
"""Handles a HTTP request that has matched a script handler.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler that matched.
match: A re.MatchObject containing the result of the matched URL pattern.
inst: The Instance to send the request to. If None then an appropriate
Instance will be chosen.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
raise NotImplementedError()
def handle(self, match, environ, start_response):
"""Serves the file content matching the request.
Args:
match: The re.MatchObject containing the result of matching the URL
against this handler's URL pattern.
environ: An environ dict for the current request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
relative_path = match.expand(self._url_map.static_files)
if not self._is_relative_path_valid(relative_path):
if self._url_map.require_matching_file:
return None
else:
return self._not_found_404(environ, start_response)
full_path = os.path.join(self._root_path, relative_path)
return self._handle_path(full_path, environ, start_response)
def _search(self, string):
"""
Takes a string and returns a re.MatchObject for the Parser's regex.
"""
# TODO(LH): catch exceptions for malformed regex
return re.search(self.regex, string, re.DOTALL | re.IGNORECASE)
def _handle_instance_request(self,
environ,
start_response,
url_map,
match,
request_id,
inst,
request_type):
"""Handles a request routed a particular Instance.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler that matched.
match: A re.MatchObject containing the result of the matched URL pattern.
request_id: A unique string id associated with the request.
inst: The instance.Instance to send the request to.
request_type: The type of the request. See instance.*_REQUEST module
constants.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
if request_type != instance.READY_REQUEST:
with self._condition:
self._num_outstanding_instance_requests += 1
self._outstanding_request_history.append(
(time.time(), self.num_outstanding_instance_requests))
try:
logging.debug('Dispatching request to %s', inst)
return inst.handle(environ, start_response, url_map, match, request_id,
request_type)
finally:
with self._condition:
if request_type != instance.READY_REQUEST:
self._num_outstanding_instance_requests -= 1
self._condition.notify()
def handle(self, environ, start_response, url_map, match, request_id,
request_type):
"""Serves a request by displaying an error page.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler matching this request.
match: A re.MatchObject containing the result of the matched URL pattern.
request_id: A unique string id associated with the request.
request_type: The type of the request. See instance.*_REQUEST module
constants.
Yields:
A sequence of strings containing the body of the HTTP response.
"""
start_response('500 Internal Server Error',
[('Content-Type', 'text/html')])
yield '<html><head><title>Invalid PHP Configuration</title></head>'
yield '<body>'
yield '<title>Invalid PHP Configuration</title>'
if isinstance(self._exception, _PHPEnvironmentError):
yield '<b>The PHP interpreter specified with the --php_executable_path '
yield ' flag ("%s") is not compatible with the App Engine ' % (
self._php_executable_path)
yield 'PHP development environment.</b><br>'
yield '<br>'
yield '<pre>%s</pre>' % self._exception
else:
yield '<b>%s</b>' % cgi.escape(str(self._exception))
yield '</body></html>'
def match_url_regex(rules, url, callback):
"""Given rules and a callback, find the rule that matches the url.
Rules look like::
(
{
'schemes': ['https', 'ssh'],
'netlocs': ['hg.mozilla.org'],
'path_regexes': [
"^(?P<path>/mozilla-(central|unified))(/|$)",
]
},
...
)
Args:
rules (list): a list of dictionaries specifying lists of ``schemes``,
``netlocs``, and ``path_regexes``.
url (str): the url to test
callback (function): a callback that takes an ``re.MatchObject``.
If it returns None, continue searching. Otherwise, return the
value from the callback.
Returns:
value: the value from the callback, or None if no match.
"""
parts = urlparse(url)
path = unquote(parts.path)
for rule in rules:
if parts.scheme not in rule['schemes']:
continue
if parts.netloc not in rule['netlocs']:
continue
for regex in rule['path_regexes']:
m = re.search(regex, path)
if m is None:
continue
result = callback(m)
if result is not None:
return result
def _handle_instance_request(self,
environ,
start_response,
url_map,
match,
request_id,
inst,
request_type):
"""Handles a request routed a particular Instance.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler that matched.
match: A re.MatchObject containing the result of the matched URL pattern.
request_id: A unique string id associated with the request.
inst: The instance.Instance to send the request to.
request_type: The type of the request. See instance.*_REQUEST module
constants.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
start_time = time.time()
timeout_time = start_time + self._get_wait_time()
try:
while time.time() < timeout_time:
logging.debug('Dispatching request to %s after %0.4fs pending',
inst, time.time() - start_time)
try:
return inst.handle(environ, start_response, url_map, match,
request_id, request_type)
except instance.CannotAcceptRequests:
pass
inst.wait(timeout_time)
if inst.has_quit:
return self._error_response(environ, start_response, 503)
else:
return self._error_response(environ, start_response, 503)
finally:
with self._condition:
self._condition.notify()
def _handle_script_request(self,
environ,
start_response,
url_map,
match,
request_id,
inst=None,
request_type=instance.NORMAL_REQUEST):
"""Handles a HTTP request that has matched a script handler.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler that matched.
match: A re.MatchObject containing the result of the matched URL pattern.
request_id: A unique string id associated with the request.
inst: The instance.Instance to send the request to. If None then an
appropriate instance.Instance will be chosen.
request_type: The type of the request. See instance.*_REQUEST module
constants.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
if ((request_type in (instance.NORMAL_REQUEST, instance.READY_REQUEST) and
self._suspended) or self._quit_event.is_set()):
return self._error_response(environ, start_response, 404)
if self._module_configuration.is_backend:
environ['BACKEND_ID'] = self._module_configuration.module_name
else:
environ['BACKEND_ID'] = (
self._module_configuration.version_id.split('.', 1)[0])
if inst is not None:
return self._handle_instance_request(
environ, start_response, url_map, match, request_id, inst,
request_type)
start_time = time.time()
timeout_time = start_time + self._get_wait_time()
while time.time() < timeout_time:
if ((request_type in (instance.NORMAL_REQUEST, instance.READY_REQUEST) and
self._suspended) or self._quit_event.is_set()):
return self._error_response(environ, start_response, 404)
inst = self._choose_instance(timeout_time)
if inst:
try:
logging.debug('Dispatching request to %s after %0.4fs pending',
inst, time.time() - start_time)
return inst.handle(environ, start_response, url_map, match,
request_id, request_type)
except instance.CannotAcceptRequests:
continue
finally:
with self._condition:
self._condition.notify()
else:
return self._error_response(environ, start_response, 503, _TIMEOUT_HTML)
def _handle_instance_request(self,
environ,
start_response,
url_map,
match,
request_id,
inst,
request_type):
"""Handles a request routed a particular Instance.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler that matched.
match: A re.MatchObject containing the result of the matched URL pattern.
request_id: A unique string id associated with the request.
inst: The instance.Instance to send the request to.
request_type: The type of the request. See instance.*_REQUEST module
constants.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
start_time = time.time()
timeout_time = start_time + self._get_wait_time()
try:
while time.time() < timeout_time:
logging.debug('Dispatching request to %s after %0.4fs pending',
inst, time.time() - start_time)
try:
return inst.handle(environ, start_response, url_map, match,
request_id, request_type)
except instance.CannotAcceptRequests:
pass
inst.wait(timeout_time)
if inst.has_quit:
return self._error_response(environ, start_response, 503)
return self._error_response(environ, start_response, 503)
finally:
with self._condition:
self._condition.notify()
def _handle_instance_request(self,
environ,
start_response,
url_map,
match,
request_id,
inst,
request_type):
"""Handles a request routed a particular Instance.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler that matched.
match: A re.MatchObject containing the result of the matched URL pattern.
request_id: A unique string id associated with the request.
inst: The instance.Instance to send the request to.
request_type: The type of the request. See instance.*_REQUEST module
constants.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
instance_id = inst.instance_id
start_time = time.time()
timeout_time = start_time + self._get_wait_time()
try:
while time.time() < timeout_time:
logging.debug('Dispatching request to %s after %0.4fs pending',
inst, time.time() - start_time)
try:
return inst.handle(environ, start_response, url_map, match,
request_id, request_type)
except instance.CannotAcceptRequests:
pass
if inst.has_quit:
return self._error_response(environ, start_response, 503)
with self._condition:
if self._instance_running[instance_id]:
should_start = False
else:
self._instance_running[instance_id] = True
should_start = True
if should_start:
self._start_instance(instance_id)
else:
inst.wait(timeout_time)
else:
return self._error_response(environ, start_response, 503)
finally:
with self._condition:
self._condition.notify()
def _handle_script_request(self,
environ,
start_response,
url_map,
match,
request_id,
inst=None,
request_type=instance.NORMAL_REQUEST):
"""Handles a HTTP request that has matched a script handler.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler that matched.
match: A re.MatchObject containing the result of the matched URL pattern.
request_id: A unique string id associated with the request.
inst: The instance.Instance to send the request to. If None then an
appropriate instance.Instance will be chosen.
request_type: The type of the request. See instance.*_REQUEST module
constants.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
if self._quit_event.is_set():
return self._error_response(environ, start_response, 404)
if self._module_configuration.is_backend:
environ['BACKEND_ID'] = self._module_configuration.module_name
else:
environ['BACKEND_ID'] = (
self._module_configuration.version_id.split('.', 1)[0])
if inst is not None:
return self._handle_instance_request(
environ, start_response, url_map, match, request_id, inst,
request_type)
start_time = time.time()
timeout_time = start_time + self._get_wait_time()
while time.time() < timeout_time:
if self._quit_event.is_set():
return self._error_response(environ, start_response, 404)
inst = self._choose_instance(timeout_time)
if inst:
try:
logging.debug('Dispatching request to %s after %0.4fs pending',
inst, time.time() - start_time)
return inst.handle(environ, start_response, url_map, match,
request_id, request_type)
except instance.CannotAcceptRequests:
continue
finally:
with self._condition:
self._condition.notify()
else:
return self._error_response(environ, start_response, 503, _TIMEOUT_HTML)
def _handle_script_request(self,
environ,
start_response,
url_map,
match,
request_id,
inst=None,
request_type=instance.INTERACTIVE_REQUEST):
"""Handles a interactive request by forwarding it to the managed Instance.
Args:
environ: An environ dict for the request as defined in PEP-333.
start_response: A function with semantics defined in PEP-333.
url_map: An appinfo.URLMap instance containing the configuration for the
handler that matched.
match: A re.MatchObject containing the result of the matched URL pattern.
request_id: A unique string id associated with the request.
inst: The instance.Instance to send the request to.
request_type: The type of the request. See instance.*_REQUEST module
constants. This must be instance.INTERACTIVE_REQUEST.
Returns:
An iterable over strings containing the body of the HTTP response.
"""
assert inst is None
assert request_type == instance.INTERACTIVE_REQUEST
start_time = time.time()
timeout_time = start_time + self._get_wait_time()
while time.time() < timeout_time:
new_instance = False
with self._inst_lock:
if not self._inst:
self._inst = self._instance_factory.new_instance(
AutoScalingModule.generate_instance_id(),
expect_ready_request=False)
new_instance = True
inst = self._inst
if new_instance:
self._inst.start()
try:
return inst.handle(environ, start_response, url_map, match,
request_id, request_type)
except instance.CannotAcceptRequests:
inst.wait(timeout_time)
except Exception:
# If the instance is restarted while handling a request then the
# exception raises is unpredictable.
if inst != self._inst:
start_response('503 Service Unavailable', [])
return ['Instance was restarted while executing command']
logging.exception('Unexpected exception handling command: %r', environ)
raise
else:
start_response('503 Service Unavailable', [])
return ['The command timed-out while waiting for another one to complete']
def _read_until(term, pattern, timeout):
"""
Convenience read-until-pattern function, supporting :meth:`~.get_location`.
:arg blessed.Terminal term: :class:`~.Terminal` instance.
:arg float timeout: timeout period, may be set to None to indicate no
timeout (where 0 is always returned).
:arg str pattern: target regular expression pattern to seek.
:rtype: tuple
:returns: tuple in form of ``(match, str)``, *match*
may be :class:`re.MatchObject` if pattern is discovered
in input stream before timeout has elapsed, otherwise
None. ``str`` is any remaining text received exclusive
of the matching pattern).
The reason a tuple containing non-matching data is returned, is that the
consumer should push such data back into the input buffer by
:meth:`~.Terminal.ungetch` if any was received.
For example, when a user is performing rapid input keystrokes while its
terminal emulator surreptitiously responds to this in-band sequence, we
must ensure any such keyboard data is well-received by the next call to
term.inkey() without delay.
"""
stime = time.time()
match, buf = None, u''
# first, buffer all pending data. pexpect library provides a
# 'searchwindowsize' attribute that limits this memory region. We're not
# concerned about OOM conditions: only (human) keyboard input and terminal
# response sequences are expected.
while True:
# block as long as necessary to ensure at least one character is
# received on input or remaining timeout has elapsed.
ucs = term.inkey(timeout=_time_left(stime, timeout))
if ucs:
buf += ucs
# while the keyboard buffer is "hot" (has input), we continue to
# aggregate all awaiting data. We do this to ensure slow I/O
# calls do not unnecessarily give up within the first 'while' loop
# for short timeout periods.
while True:
ucs = term.inkey(timeout=0)
if not ucs:
break
buf += ucs
match = re.search(pattern=pattern, string=buf)
if match is not None:
# match
break
if timeout is not None:
if not _time_left(stime, timeout):
# timeout
break
return match, buf