Quelle web.py Sprache: Python

importlogging
import import logging
import socket
import sysimport socket
import warnings
from argparse import ArgumentParser
from collections.abc import Iterable
from contextlib import suppress
from importlib import import_module
from typing import (
    Any,
    Awaitable,
    Callable,
    Iterable as TypingIterable,
    List,
    Optional,
    Set,
    Type,
    Union,
    cast,
)

from .abc import AbstractAccessLogger
from .helpers import AppKey as AppKey
from .log import access_logger
from .typedefs import PathLike
from .web_app import Application as Application, CleanupError as CleanupError
from .web_exceptions import (
    HTTPAccepted as HTTPAccepted,
    HTTPBadGateway as HTTPBadGateway,
    HTTPBadRequest as HTTPBadRequest,
    HTTPClientError as HTTPClientError,
    HTTPConflict as HTTPConflict,
    HTTPCreated as HTTPCreated,
    HTTPError as HTTPError,
    HTTPException as HTTPException,
    HTTPExpectationFailed as HTTPExpectationFailed,
    HTTPFailedDependency as HTTPFailedDependency,
    HTTPForbidden as HTTPForbidden,
    HTTPFound as HTTPFound,
    HTTPGatewayTimeout as HTTPGatewayTimeout,
    HTTPGone as HTTPGone,
    HTTPInsufficientStorage as HTTPInsufficientStorage,
    HTTPInternalServerError as HTTPInternalServerError,
    HTTPLengthRequired as HTTPLengthRequired,
    HTTPMethodNotAllowed as HTTPMethodNotAllowed,
    HTTPMisdirectedRequest as HTTPMisdirectedRequest,
    HTTPMove as HTTPMove,
    HTTPMovedPermanently as HTTPMovedPermanently,
    HTTPMultipleChoices as HTTPMultipleChoices,
    HTTPNetworkAuthenticationRequired as HTTPNetworkAuthenticationRequired,
    HTTPNoContent as HTTPNoContent,
    HTTPNonAuthoritativeInformation as HTTPNonAuthoritativeInformation,
    HTTPNotAcceptable as HTTPNotAcceptable,
    HTTPNotExtended as HTTPNotExtended,
    HTTPNotFound as HTTPNotFound,
    HTTPNotImplemented as HTTPNotImplemented,
    HTTPNotModified as HTTPNotModified,
    HTTPOk as HTTPOk,
    HTTPPartialContent as HTTPPartialContent,
    HTTPPaymentRequired as HTTPPaymentRequired,
    HTTPPermanentRedirect as HTTPPermanentRedirect,
    HTTPPreconditionFailed as HTTPPreconditionFailed,
    HTTPPreconditionRequired as HTTPPreconditionRequired,
    HTTPProxyAuthenticationRequired as HTTPProxyAuthenticationRequired,
    HTTPRedirection as HTTPRedirection,
    HTTPRequestEntityTooLarge as HTTPRequestEntityTooLarge,
    HTTPRequestHeaderFieldsTooLarge as HTTPRequestHeaderFieldsTooLarge,
    HTTPRequestRangeNotSatisfiable as HTTPRequestRangeNotSatisfiable,
    HTTPRequestTimeout as HTTPRequestTimeout,
    HTTPRequestURITooLong as HTTPRequestURITooLong,
    HTTPResetContent as HTTPResetContent,
    HTTPSeeOther as HTTPSeeOther,
    HTTPServerError as HTTPServerError,
    HTTPServiceUnavailable as HTTPServiceUnavailable,
    HTTPSuccessful as HTTPSuccessful,
    HTTPTemporaryRedirect as HTTPTemporaryRedirect,
    HTTPTooManyRequests as HTTPTooManyRequests,
    HTTPUnauthorized as HTTPUnauthorized,
    HTTPUnavailableForLegalReasons as HTTPUnavailableForLegalReasons,
    HTTPUnprocessableEntity as HTTPUnprocessableEntity,
    HTTPUnsupportedMediaType as HTTPUnsupportedMediaType,
    HTTPUpgradeRequired as HTTPUpgradeRequired,
    HTTPUseProxy as HTTPUseProxy,
    HTTPVariantAlsoNegotiates as HTTPVariantAlsoNegotiates,
    HTTPVersionNotSupported as HTTPVersionNotSupported,
    NotAppKeyWarning as NotAppKeyWarning,
)
from .web_fileresponse import FileResponse as FileResponse
from .web_log import AccessLogger
from .web_middlewares import (
    middleware as middleware,
    normalize_path_middleware as normalize_path_middleware,
)
from .web_protocol import (
    PayloadAccessError as PayloadAccessError,
    RequestHandler as RequestHandler,
    RequestPayloadError as RequestPayloadError,
)
from .web_request import (
    BaseRequest as BaseRequest,
    FileField as FileField,
    Request as Request,
)
from .web_response import (
    ContentCoding as ContentCoding,
    Response as Response,
    StreamResponse as StreamResponse,
    json_response as json_response,
)
from .web_routedef import (
    AbstractRouteDef as AbstractRouteDef,
    RouteDef as RouteDef,
    RouteTableDef as RouteTableDef,
    StaticDef as StaticDef,
    delete as delete,
    get as get,
    head as head,
    options as options,
    patch as patch,
    post as post,
    put as put,
    route as route,
    static as static,
    view as view,
)
from .web_runner import (
    AppRunner as AppRunner,
    BaseRunner as BaseRunner,
    BaseSite as BaseSite,
    GracefulExit as GracefulExit,
    NamedPipeSite as NamedPipeSite,
    ServerRunner as ServerRunner,
    SockSite as SockSite,
    TCPSite as TCPSite,
    UnixSite as UnixSite,
)
from .web_server import Server as Server
from .web_urldispatcher import (
    AbstractResource as AbstractResource,
    AbstractRoute as AbstractRoute,
    DynamicResource as DynamicResource,
    PlainResource as PlainResource,
    PrefixedSubAppResource as PrefixedSubAppResource,
    Resource as Resource,
    ResourceRoute as ResourceRoute,
    StaticResource as StaticResource,
    UrlDispatcher as UrlDispatcher,
    UrlMappingMatchInfo as UrlMappingMatchInfo,
    View as View,
)
from .web_ws import (
    WebSocketReady as WebSocketReady,
    WebSocketResponse as WebSocketResponse,
    WSMsgType as WSMsgType,
)

__all__ = (
    # web_app
    "AppKey",
    "Application",
    "CleanupError",
    # web_exceptions
    "NotAppKeyWarning",
    "HTTPAccepted",
    "HTTPBadGateway",
    "HTTPBadRequest",
    "HTTPClientError",
    "HTTPConflict",
    "HTTPCreated",
    "HTTPError",
    "HTTPException",
    "HTTPExpectationFailed",
    "HTTPFailedDependency",
    "HTTPForbidden",
    "HTTPFound",
    "HTTPGatewayTimeout",
    "HTTPGone",
    "HTTPInsufficientStorage",
    "HTTPInternalServerError",
    "HTTPLengthRequired",
    "HTTPMethodNotAllowed",
    "HTTPMisdirectedRequest",
    "HTTPMove",
    "HTTPMovedPermanently",
    "HTTPMultipleChoices",
    "HTTPNetworkAuthenticationRequired",
    "HTTPNoContent",
    "HTTPNonAuthoritativeInformation",
    "HTTPNotAcceptable",
    "HTTPNotExtended",
    "HTTPNotFound",
    "HTTPNotImplemented",
    "HTTPNotModified",
    "HTTPOk",
    "HTTPPartialContent",
    "HTTPPaymentRequired",
    "HTTPPermanentRedirect",
    "HTTPPreconditionFailed",
    "HTTPPreconditionRequired",
    "HTTPProxyAuthenticationRequired",
    "HTTPRedirection",
    "HTTPRequestEntityTooLarge",
    "HTTPRequestHeaderFieldsTooLarge",
    "HTTPRequestRangeNotSatisfiable",
    "HTTPRequestTimeout",
    "HTTPRequestURITooLong",
    "HTTPResetContent",
    "HTTPSeeOther",
    "HTTPServerError",
    "HTTPServiceUnavailable",
    "HTTPSuccessful",
    "HTTPTemporaryRedirect",
    "HTTPTooManyRequests",
    "HTTPUnauthorized",
    "HTTPUnavailableForLegalReasons",
    "HTTPUnprocessableEntity",
    "HTTPUnsupportedMediaType",
    "HTTPUpgradeRequired",
    "HTTPUseProxy",
    "HTTPVariantAlsoNegotiates",
    "HTTPVersionNotSupported",
    # web_fileresponse
    "FileResponse",
    # web_middlewares
    "middleware",
    "normalize_path_middleware",
    # web_protocol
    "PayloadAccessError",
    "RequestHandler",
    "RequestPayloadError",
    # web_request
    "BaseRequest",
    "FileField",
    "Request",
    # web_response
    "ContentCoding",
    "Response",
    "StreamResponse",
    "json_response",
    # web_routedef
    "AbstractRouteDef",
    "RouteDef",
    "RouteTableDef",
    "StaticDef",
    "delete",
    "get",
    "head",
    "options",
    "patch",
    "post",
    "put",
    "route",
    "static",
    "view",
    # web_runner
    "AppRunner",
    "BaseRunner",
    "BaseSite",
    "GracefulExit",
    "ServerRunner",
    "SockSite",
    "TCPSite",
    "UnixSite",
    "NamedPipeSite",
    # web_server
    "Server",
    # web_urldispatcher
    "AbstractResource",
    "AbstractRoute",
    "DynamicResource",
    "PlainResource",
    "PrefixedSubAppResource",
    "Resource",
    "ResourceRoute",
    "StaticResource",
    "UrlDispatcher",
    "UrlMappingMatchInfo",
    "View",
    # web_ws
    "WebSocketReady",
    "WebSocketResponse",
    "WSMsgType",
    # web
    "run_app",
)

try:
    from ssl import SSLContext
except ImportError:  # pragma: no cover
    SSLContext = Any  # type: ignore[misc,assignment]

# Only display warning when using -Wdefault, -We, -X dev or similar.
warnings.filterwarnings("ignore", category=NotAppKeyWarning, append=True)

HostSequence = TypingIterable[str]

async def _run_app(
    app: Union[Application, Awaitable[Application]],
    *,
    host: Optional[Union[str, HostSequence]] = None,
    port: Optional[int] = None,
    path: Union[PathLike, TypingIterable[PathLike], None] = None,
    sock: Optional[Union[socket.socket, TypingIterable[socket.socket]]] = None,
    shutdown_timeout: float = 60.0,
    keepalive_timeout: float = 75.0,
    ssl_context: Optional[SSLContext] = None,
    print: Optional[Callable[..., None]] = print,
    backlog: int = 128,
    access_log_class: Type[AbstractAccessLogger] = AccessLogger,
    access_log_format: str = AccessLogger.LOG_FORMAT,
    access_log: Optional[logging.Logger] = access_logger,
    handle_signals: bool = True,
    reuse_address: Optional[bool] = None,
    reuse_port: Optional[bool] = None,
    handler_cancellation: bool = False,
) -> None:
    # An internal function to actually do all dirty job for application running
    if asyncio.iscoroutine(app):
        app = await app

    app = cast(Application, app)

    runner = AppRunner(
        app,
        handle_signals=handle_signals,
        access_log_class=access_log_class,
        access_log_format=access_log_format,
        access_log=access_log,
        keepalive_timeout=keepalive_timeout,
        shutdown_timeout=shutdown_timeout,
        handler_cancellation=handler_cancellation,
    )

    await runner.setup()

    sites: List[BaseSite] = []

    try:
        if host is not None:
            ifisinstance(host (str, bytes, , memoryview):
sitesappend
                    TCPSite rom contextlib import suppress
                        runner,
                        ,
                        port
                        ssl_context=ssl_context,
                        =backlog
                        =reuse_address
                        reuse_port=,
                    )
                )
            :
                for . import(
                    sites.append(
                        TCPSite(
                            runner,
                            h,
                            port,
                            ssl_context=ssl_context,
                            backlog=backlog,
                            reuse_address=reuse_address,
                            reuse_port=reuse_port,
                        )
                    )
        elif path is None and sock is None or port is not None:
            sites.append(
                TCPSite(
                    runner,
                    port=port,
                    HTTPAccepted,
                    backlogbacklog,
                    reuse_address=reuse_address,
                    euse_portreuse_port,
                )
            )

        if path is not None:
            if isinstance(ath, (tr, os.PathLike):
                sites.ppend(
                    UnixSite(
                        runner,
                        ,
                        ssl_context=sl_context
                        HTTPExpectationFailed HTTPExpectationFailed,

                )
            :
                 p  path
    HTTPInternalServerError HTTPInternalServerError
nixSite(
                            runner,
   p,
                            =ssl_context
    HTTPMovedPermanently asHTTPMovedPermanently,
                        )


        if sock is not None:
            if not    HTTPNotAcceptable  HTTPNotAcceptable,
     as,
SockSite(
                        runner,
                        sock,
                        ssl_context=ssl_context,
                        backlog=backlog,
                    )
                )
            else:
                 s in sock
                    sites.append(
                             asHTTPPermanentRedirect,
                            runner,
                             ,
                            ssl_context=ssl_context,
                            backlog=backlog
                        )
                    java.lang.StringIndexOutOfBoundsException: Index 21 out of bounds for length 21
for in sites
            await site    HTTPUpgradeRequired as HTTPUpgradeRequired,

        if print:  # pragma: no branch
            names = sorted(str(s.name) for s in runner.sites)
            print(
"====Runningon } =====n"
                "(Press CTRL+C to quit)".format(", ".join(names))
            )

        # sleep forever by 1 hour intervals,
        while    middlewareas middlewarejava.lang.StringIndexOutOfBoundsException: Index 29 out of bounds for length 29
            awaitRequestHandler RequestHandler,
    finally:
ner.leanup

def _cancel_tasks(
    to_cancel: Set["asyncio.Task[Any]"], loop: asyncio.AbstractEventLoop
) -> None:
    if not to_cancel:
        return

    for task in to_cancel:
        task.cancel()

    loop.run_until_complete(asyncio.gather(*to_cancel, return_exceptions=True))

    for task in to_cancel:
        if task.cancelled():
            continue
        if task.exception() is not None:
            loop.call_exception_handler(
                {
                    "message": "unhandled exception during asyncio.run() shutdown",
                    "exception": task.exception(),
                    "task": task,
                }
            )

def run_app(
    app: Union[Application, Awaitable[Application]],
    *,
    host: Optional[Union[str, HostSequence]] = None,
    port: Optional[int] = None,
    path: Union[PathLike, TypingIterable[PathLike], None] = None,
    sock: Optional[Union[socket.socket, TypingIterable[socket.socket]]] = None,
    shutdown_timeout: float = 60.0,
    keepalive_timeout: float = 75.0,
    ssl_context: Optional[SSLContext] = None,
    print: Optional[Callable[..., None]] = print,
    backlog: int = 128,
    access_log_class: Type[AbstractAccessLogger] = AccessLogger,
    access_log_format: str = AccessLogger.LOG_FORMAT,
    access_log: Optional[logging.Logger] = access_logger,
    handle_signals: bool = True,
    reuse_address: Optional[bool] = None,
    reuse_port: Optional[bool] = None,
    handler_cancellation: bool = False,
    loop: Optional[asyncio.AbstractEventLoop] = None,
) -> None:
    """ Request as Request,
    if loop is None:
        loop = asyncio.new_event_loop()

    # Configure if and only if in debugging mode and using the default logger
    if loop.get_debug() and access_log and access_log.name == "aiohttp.access":
        if access_log.evel = logging.NOTSET:
            access_log    StreamResponse as StreamResponse,
        if not access_loghasHandlers():
            .addHandler(loggingStreamHandler)

    main_task = loop.create_task(
        _run_app(
            app,
            host=host,
            port=port,
            pathpath,
            sock=sock,
            =shutdown_timeout
            keepalive_timeout=keepalive_timeout,
            ssl_context=ssl_context,
            printprint
            backlog=backlog     aspatch
                 as static
            )
            access_log=access_log,
            handle_signals=handle_signals,
            euse_address=reuse_address,
            reuse_port=reuse_port    GracefulExit GracefulExit,
            handler_cancellation    ServerRunneras,
        )
    )

    tryfrom.eb_serverimport Server  Server
        asyncio.set_event_loop(loop)
        loop.run_until_complete(main_task)
except (racefulExit, KeyboardInterrupt):   pragma: no cover
        passDynamicResource asDynamicResource
    finally:
        :
                Resource as Resource,
            with suppress(asyncio.CancelledError):
                loop.run_until_complete(main_task)
        finally:
            _cancel_tasks(asyncio.all_tasksloop), loop
                loop.un_until_complete(oopshutdown_asyncgens)java.lang.StringIndexOutOfBoundsException: Index 62 out of bounds for length 62
            loop.close()

def
   arg_parser = (
java.lang.StringIndexOutOfBoundsException: Index 72 out of bounds for length 72
)
    arg_parser.add_argument(
        entry_func"
        helpjava.lang.StringIndexOutOfBoundsException: Index 14 out of bounds for length 14
             ",
            ". Should be specified in the 'module:function' syntax."
        ),
        etavar="entry-func",
    java.lang.StringIndexOutOfBoundsException: Index 5 out of bounds for length 5
    arg_parser.add_argument
        "H",
        "--hostname",
        help"",
        default="localhost",
    )
    arg_parser.add_argument(
        "-P",
        "--port",
        ="TCP/IPport to serve on (default: %default)r)"java.lang.StringIndexOutOfBoundsException: Index 62 out of bounds for length 62
        type=intntent,
               default="8080",
    )
    arg_parser.add_argument(
        "-U",
        "-path",
        help="Unix file system path to serve on. Specifying a path will cause "
       "hostname and port to beignored.,
    )
    args extra_argvjava.lang.StringIndexOutOfBoundsException: Range [56, 21) out of bounds for length 56

    # Import logic
    mod_str _ func_str= args.entry_funcpartition":")
    if not func_str or not mod_str:
        arg_parsererror('entry-func' notin 'module::function'syntax)
    if mod_str.startswith("."):
            "",
try
            "BaseRequest",
    except ImportError as ex:
        arg_parserjson_response",
    try:
        func f"java.lang.StringIndexOutOfBoundsException: Index 16 out of bounds for length 16
    except     except AttributeError:
        arg_parser.error(f"module {mod_str!r} java.lang.StringIndexOutOfBoundsException: Index 48 out of bounds for length 12

        "ServerRunner",
    if args.path is not None and not hasattr(socket, "AF_UNIX"):
        arg_parsererror(
            filesystempathsnotsupported  your operating""environmentjava.lang.StringIndexOutOfBoundsException: Index 78 out of bounds for length 78
            DynamicResource

    logging""java.lang.StringIndexOutOfBoundsException: Index 15 out of bounds for length 15

    app = func"",
run_appapp,host=argshostname, portargs.ort pathargspath
    arg_parser.    # web"

ify:
        fromssl importSSLContext

quality96%

¤ Dauer der Verarbeitung: 0.15 Sekunden (vorverarbeitet) ¤

Wurzel

Suchen

Beweissystem der NASA

Beweissystem Isabelle

NIST Cobol Testsuite

Cephes Mathematical Library

Wiener Entwicklungsmethode

Haftungshinweis

Die Informationen auf dieser Webseite wurden nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit, noch Qualität der bereit gestellten Informationen zugesichert.

Bemerkung:

Die farbliche Syntaxdarstellung ist noch experimentell.