Search Options

Results per page
Sort
Preferred Languages
Advance

Results 1 - 10 of 59 for state (0.2 sec)

  1. tests/test_dependency_contextmanager.py

    class OtherDependencyError(Exception):
        pass
    
    
    async def asyncgen_state(state: Dict[str, str] = Depends(get_state)):
        state["/async"] = "asyncgen started"
        yield state["/async"]
        state["/async"] = "asyncgen completed"
    
    
    def generator_state(state: Dict[str, str] = Depends(get_state)):
        state["/sync"] = "generator started"
        yield state["/sync"]
        state["/sync"] = "generator completed"
    
    
    Python
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Sat Feb 24 23:06:37 GMT 2024
    - 11.6K bytes
    - Viewed (0)
  2. docs_src/sql_databases_peewee/sql_app/main.py

    from .database import db_state_default
    
    database.db.connect()
    database.db.create_tables([models.User, models.Item])
    database.db.close()
    
    app = FastAPI()
    
    sleep_time = 10
    
    
    async def reset_db_state():
        database.db._state._state.set(db_state_default.copy())
        database.db._state.reset()
    
    
    def get_db(db_state=Depends(reset_db_state)):
        try:
            database.db.connect()
    Python
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Thu Mar 26 19:09:53 GMT 2020
    - 2.2K bytes
    - Viewed (0)
  3. docs/en/docs/how-to/sql-databases-peewee.md

    ```Python
    # db._state = PeeweeConnectionState()
    ```
    
    And in the file `sql_app/main.py` file, comment the body of the `async` dependency `reset_db_state()` and replace it with a `pass`:
    
    ```Python
    async def reset_db_state():
    #     database.db._state._state.set(db_state_default.copy())
    #     database.db._state.reset()
        pass
    ```
    
    Then run your app with Uvicorn:
    
    Plain Text
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Tue Jan 16 13:23:25 GMT 2024
    - 23.6K bytes
    - Viewed (0)
  4. tests/test_router_events.py

        app.include_router(router)
    
        assert state.app_startup is False
        assert state.router_startup is False
        assert state.sub_router_startup is False
        assert state.app_shutdown is False
        assert state.router_shutdown is False
        assert state.sub_router_shutdown is False
        with TestClient(app) as client:
            assert state.app_startup is True
            assert state.router_startup is True
    Python
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Wed Oct 18 12:36:40 GMT 2023
    - 3.2K bytes
    - Viewed (0)
  5. tests/test_dependency_normal_exceptions.py

        return {"message": "OK"}
    
    
    @pytest.fixture(autouse=True)
    def reset_state_and_db():
        global fake_database
        global state
        fake_database = initial_fake_database.copy()
        state = initial_state.copy()
    
    
    client = TestClient(app)
    
    
    def test_dependency_gets_exception():
        assert state["except"] is False
        assert state["finally"] is False
        response = client.put("/invalid-user/rick", json="Morty")
    Python
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Sat Feb 24 23:06:37 GMT 2024
    - 1.9K bytes
    - Viewed (0)
  6. docs_src/sql_databases/sql_app_py39/alt_main.py

        response = Response("Internal server error", status_code=500)
        try:
            request.state.db = SessionLocal()
            response = await call_next(request)
        finally:
            request.state.db.close()
        return response
    
    
    # Dependency
    def get_db(request: Request):
        return request.state.db
    
    
    @app.post("/users/", response_model=schemas.User)
    Python
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Fri Jan 07 14:11:31 GMT 2022
    - 1.9K bytes
    - Viewed (0)
  7. docs/de/docs/reference/websockets.md

                - url
                - base_url
                - headers
                - query_params
                - path_params
                - cookies
                - client
                - state
                - url_for
                - client_state
                - application_state
                - receive
                - send
                - accept
                - receive_text
                - receive_bytes
                - receive_json
                - iter_text
    Plain Text
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Sat Mar 30 18:16:27 GMT 2024
    - 1.8K bytes
    - Viewed (0)
  8. docs_src/sql_databases_peewee/sql_app/database.py

    from contextvars import ContextVar
    
    import peewee
    
    DATABASE_NAME = "test.db"
    db_state_default = {"closed": None, "conn": None, "ctx": None, "transactions": None}
    db_state = ContextVar("db_state", default=db_state_default.copy())
    
    
    class PeeweeConnectionState(peewee._ConnectionState):
        def __init__(self, **kwargs):
            super().__setattr__("_state", db_state)
            super().__init__(**kwargs)
    
        def __setattr__(self, name, value):
    Python
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Thu Mar 26 19:09:53 GMT 2020
    - 662 bytes
    - Viewed (0)
  9. tests/test_dependency_contextvars.py

    legacy_request_state_context_var: ContextVar[Optional[Dict[str, Any]]] = ContextVar(
        "legacy_request_state_context_var", default=None
    )
    
    app = FastAPI()
    
    
    async def set_up_request_state_dependency():
        request_state = {"user": "deadpond"}
        contextvar_token = legacy_request_state_context_var.set(request_state)
        yield request_state
        legacy_request_state_context_var.reset(contextvar_token)
    
    
    Python
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Thu Feb 17 12:40:12 GMT 2022
    - 1.5K bytes
    - Viewed (0)
  10. docs_src/sql_databases/sql_app_py310/alt_main.py

        response = Response("Internal server error", status_code=500)
        try:
            request.state.db = SessionLocal()
            response = await call_next(request)
        finally:
            request.state.db.close()
        return response
    
    
    # Dependency
    def get_db(request: Request):
        return request.state.db
    
    
    @app.post("/users/", response_model=schemas.User)
    Python
    - Registered: Sun May 05 07:19:11 GMT 2024
    - Last Modified: Fri Jan 07 14:11:31 GMT 2022
    - 1.9K bytes
    - Viewed (0)
Back to top