我们从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