- Sort Score
- Result 10 results
- Languages All
Results 1 - 10 of 42 for wait (0.14 sec)
-
docs_src/nosql_databases/tutorial001.py
from typing import Union from couchbase import LOCKMODE_WAIT from couchbase.bucket import Bucket from couchbase.cluster import Cluster, PasswordAuthenticator from fastapi import FastAPI from pydantic import BaseModel USERPROFILE_DOC_TYPE = "userprofile" def get_bucket(): cluster = Cluster( "couchbase://couchbasehost:8091?fetch_mutation_tokens=1&operation_timeout=30&n1ql_timeout=300" )
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Sat May 14 11:59:59 GMT 2022 - 1.3K bytes - Viewed (0) -
tests/test_ws_router.py
await websocket.accept() await websocket.send_text(pathparam) await websocket.send_text(queryparam) await websocket.close() async def ws_dependency(): return "Socket Dependency" @router.websocket("/router-ws-depends/") async def router_ws_decorator_depends( websocket: WebSocket, data=Depends(ws_dependency) ): await websocket.accept() await websocket.send_text(data)
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Sun Jun 11 19:08:14 GMT 2023 - 7.5K bytes - Viewed (0) -
docs_src/custom_request_and_route/tutorial002.py
original_route_handler = super().get_route_handler() async def custom_route_handler(request: Request) -> Response: try: return await original_route_handler(request) except RequestValidationError as exc: body = await request.body() detail = {"errors": exc.errors(), "body": body.decode()} raise HTTPException(status_code=422, detail=detail)
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Fri May 13 23:38:22 GMT 2022 - 932 bytes - Viewed (0) -
docs_src/async_tests/test_main.py
import pytest from httpx import AsyncClient from .main import app @pytest.mark.anyio async def test_root(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/") assert response.status_code == 200
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Wed Oct 06 15:32:11 GMT 2021 - 306 bytes - Viewed (0) -
tests/test_dependency_contextvars.py
legacy_request_state_context_var.reset(contextvar_token) @app.middleware("http") async def custom_middleware( request: Request, call_next: Callable[[Request], Awaitable[Response]] ): response = await call_next(request) response.headers["custom"] = "foo" return response @app.get("/user", dependencies=[Depends(set_up_request_state_dependency)]) def get_user():
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Thu Feb 17 12:40:12 GMT 2022 - 1.5K bytes - Viewed (0) -
docs_src/websockets/tutorial002_py310.py
cookie_or_token: str = Depends(get_cookie_or_token), ): await websocket.accept() while True: data = await websocket.receive_text() await websocket.send_text( f"Session cookie or query token value is: {cookie_or_token}" ) if q is not None: await websocket.send_text(f"Query parameter q is: {q}")
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Sat Mar 18 12:29:59 GMT 2023 - 2.7K bytes - Viewed (0) -
fastapi/dependencies/utils.py
elif is_gen_callable(call) or is_async_gen_callable(call): solved = await solve_generator( call=call, stack=async_exit_stack, sub_values=sub_values ) elif is_coroutine_callable(call): solved = await call(**sub_values) else: solved = await run_in_threadpool(call, **sub_values) if sub_dependant.name is not None:
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Tue Apr 02 02:52:56 GMT 2024 - 29.5K bytes - Viewed (0) -
fastapi/exception_handlers.py
content={"detail": jsonable_encoder(exc.errors())}, ) async def websocket_request_validation_exception_handler( websocket: WebSocket, exc: WebSocketRequestValidationError ) -> None: await websocket.close( code=WS_1008_POLICY_VIOLATION, reason=jsonable_encoder(exc.errors())
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Sun Jun 11 19:08:14 GMT 2023 - 1.3K bytes - Viewed (0) -
configure.py
proc = subprocess.Popen( [environ_cp['PYTHON_BIN_PATH'], paths[0]] + cuda_libraries, stdout=subprocess.PIPE, env=maybe_encode_env(environ_cp)) if proc.wait(): # Errors from find_cuda_config.py were sent to stderr. print('Asking for detailed CUDA configuration...\n') return False config = dict(
Python - Registered: Tue Apr 30 12:39:09 GMT 2024 - Last Modified: Mon Apr 15 18:25:36 GMT 2024 - 53.8K bytes - Viewed (0) -
tests/test_ws_dependencies.py
async def index(websocket: WebSocket, deps: DepList): await websocket.accept() await websocket.send_text(json.dumps(deps)) await websocket.close() @router.websocket("/router", dependencies=[create_dependency("routerindex")]) async def routerindex(websocket: WebSocket, deps: DepList): await websocket.accept() await websocket.send_text(json.dumps(deps)) await websocket.close()
Python - Registered: Sun Apr 28 07:19:10 GMT 2024 - Last Modified: Sun Jun 11 20:35:39 GMT 2023 - 2.1K bytes - Viewed (0)