programming python

Trình quản lý bối cảnh có điều kiện của Python

Trong những năm qua, tôi đã sử dụng contextlib.ExitStack của Python theo một số cách thú vị. Viên chức này quảng cáo nó như một cách để quản lý nhiều trình quản lý bối cảnh và có một vài ví dụ về cách tận dụng nó. Tuy nhiên, trong tài liệu cũng như trong tìm kiếm mã GitHub, tôi không thể tìm thấy các ví dụ về một số cách có thể bất thường mà tôi đã sử dụng trước đây. Vì vậy, tôi nghĩ rằng tôi sẽ ghi lại chúng ở đây

Thực thi giao dịch cấp yêu cầu

Trong khi sử dụng API, điều quan trọng là phải xử lý lỗi theo cách ngăn ngừa hỏng trạng thái cơ sở dữ liệu. Trong ví dụ sau, tôi đang thực hiện hai yêu cầu

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

0 tới một API và quay trở lại trạng thái ban đầu nếu bất kỳ yêu cầu nào trong số đó không thành công

# src.py
from __future__ import annotations

import logging
import uuid
from contextlib import ExitStack
from http import HTTPStatus

import httpx

logging.basicConfig[level=logging.INFO]


def group_create[uuid_a: str, uuid_b: str] -> tuple[httpx.Response, ...]:
    with httpx.Client[] as client:
        url = "//httpbin.org/post"
        response_a = client.post[
            url,
            json={"uuid": uuid_a, "foo": "bar"},
        ]
        response_b = client.post[
            url,
            json={"uuid": uuid_b, "fizz": "bazz"},
        ]
    return response_a, response_b


def maybe_rollback[
    uuid: str,
    incoming_status_code: int,
    expected_status_code: int = HTTPStatus.OK,
] -> None:
    if incoming_status_code != expected_status_code:
        logging.info[f"Rolling back request: {uuid}"]
        url = f"//httpbin.org/delete?uuid={uuid}"
        response = httpx.delete[url]
        assert response.status_code == HTTPStatus.OK
    else:
        logging.info[f"Request {uuid} completed successfully."]


def main[] -> None:
    with ExitStack[] as stack:
        uuid_a = str[uuid.uuid4[]]
        uuid_b = str[uuid.uuid4[]]

        response_a, response_b = group_create[uuid_a, uuid_b]
        stack.callback[
            maybe_rollback,
            uuid=uuid_a,
            incoming_status_code=response_a.status_code,
        ]
        stack.callback[
            maybe_rollback,
            uuid=uuid_b,
            incoming_status_code=response_b.status_code,
        ]


if __name__ == "__main__":
    main[]

Chạy cái này sẽ in đầu ra sau

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

Ở đây, hàm

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

1 thực hiện hai lệnh gọi đến điểm cuối

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

2 và hàm

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

3 xóa bản ghi đã tạo nếu bất kỳ một trong hai yêu cầu không thành công. Trong hàm

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

4, tôi đã sử dụng phương thức

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

5 để đăng ký cuộc gọi lại

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

3. Nếu bạn thay đổi

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

7 trong hàm

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

3 thành một cái gì đó giống như

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

9, bạn sẽ có thể thấy các cuộc gọi lại dọn dẹp đang hoạt động

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

Gọi móc sự kiện có điều kiện

Chiến lược tương tự được sử dụng trong phần trước có thể được áp dụng để gọi móc sự kiện theo điều kiện. Ví dụ: giả sử bạn muốn chạy chức năng gọi lại khi chức năng sự kiện nào đó thực thi. Tuy nhiên, bạn chỉ muốn một loại chức năng gọi lại cụ thể được thực thi tùy thuộc vào trạng thái của điều kiện hoặc đường dẫn mã của bạn. Tôi đã tìm thấy mẫu sau đây hữu ích trong trường hợp này

# src.py
from __future__ import annotations

from contextlib import ExitStack
from typing import Any


class EventHook:
    def __init__[self, event_name: str] -> None:
        self.event_name = event_name
        self.dispatch_config = {
            "success": self.on_success,
            "failure": self.on_failure,
        }

    def on_success[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def on_failure[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def __call__[self] -> Any:
        return self.dispatch_config[self.event_name][]


def successful_event[] -> None:
    print["'successful_event' executed"]


def failed_event[] -> None:
    print["'failed_event' executed"]
    1 / 0


def main[] -> None:
    success_hook = EventHook["success"]
    failure_hook = EventHook["failure"]

    with ExitStack[] as stack:
        try:
            # Run successful event and attach success hook.
            successful_event[]
            stack.callback[success_hook]

            failed_event[]
        except ZeroDivisionError:
            # When the failed even raises an error, attach failure hook.
            stack.callback[failure_hook]


if __name__ == "__main__":
    main[]

'successful_event' executed
'failed_event' executed
'failure' hook called
'success' hook called

Ở đây hook

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

0 sẽ chỉ được gọi nếu có lỗi trong đường dẫn thực thi của bạn dẫn đến một ngoại lệ

Tránh cấu trúc bối cảnh lồng nhau

Nó có thể trở nên xấu đi khá nhanh khi bạn bắt đầu sử dụng nhiều trình quản lý bối cảnh lồng nhau. Ví dụ: nếu bạn cần mở hai tệp và sao chép nội dung từ tệp này sang tệp kia, thông thường bạn sẽ khởi động hai trình quản lý ngữ cảnh lồng nhau và chuyển nội dung như thế này

# src.py
with open["file1.md"] as f1:
    with open["file2.md"] as f2:
        # Copy content from f1 to f2 and save it.

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

1 có thể giúp bạn thoát khỏi chỉ với một cấp độ làm tổ tại đây. Đây là một ví dụ hoàn chỉnh

# src.py
import io
import shutil
import tempfile
from contextlib import ExitStack


def copy_over[
    fsrc: io.IOBase,
    fdst: io.IOBase,
    skip_line: int = 0,
] -> None:
    if skip_line > 0:
        for _ in range[skip_line]:
            fsrc.readline[]
    shutil.copyfileobj[fsrc, fdst]


def main[] -> None:
    with ExitStack[] as stack:
        # Enter into the respective context managers without explicit
        # 'with' blocks.
        fsrc = stack.enter_context[
            tempfile.SpooledTemporaryFile[mode="rb"],
        ]
        fdst = stack.enter_context[
            tempfile.SpooledTemporaryFile[mode="rb+"],
        ]

        # Write some data to the source file.
        fsrc.write[b"hello world\nhello mars"]

        # Rewind the source file and copy it to the destination file.
        fsrc.seek[0]
        copy_over[fsrc, fdst, skip_line=1]

        # Rewind the destination file and assert the data.
        fdst.seek[0]
        assert fdst.read[] == b"hello mars"

        # Rewind the dst file and print out the shape of the fdst
        # content.
        fdst.seek[0]
        print[fdst.read[]]


if __name__ == "__main__":
    main[]

Ví dụ này tạo hai trường hợp tệp tạm thời trong bộ nhớ với ____12_______2.

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

3 có thể được sử dụng làm trình quản lý bối cảnh. Tuy nhiên, thay vì lồng hai thể hiện, tôi đang sử dụng phương pháp

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

4 để nhập vào trình quản lý ngữ cảnh mà không sử dụng câu lệnh

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

5 một cách rõ ràng. Phương thức

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

6 này đảm bảo rằng phương thức

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

7 của trình quản lý ngữ cảnh tương ứng sẽ được gọi đúng khi kết thúc quá trình chạy hàm

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

Sau đó, trong phần nội dung của

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

1, chúng tôi đang viết một số nội dung vào tệp trong bộ nhớ đầu tiên và sau đó sao chép nội dung vào tệp trong bộ nhớ khác. Nếu chúng tôi phải mở và quản lý nhiều trình quản lý bối cảnh hơn nữa, thì theo cách này, chúng tôi có thể thực hiện điều đó mà không cần tạo thêm bất kỳ tổ hợp nào

Áp dụng nhiều bản vá làm trình quản lý ngữ cảnh

# src.py
from __future__ import annotations

from contextlib import ExitStack
from typing import Any


class EventHook:
    def __init__[self, event_name: str] -> None:
        self.event_name = event_name
        self.dispatch_config = {
            "success": self.on_success,
            "failure": self.on_failure,
        }

    def on_success[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def on_failure[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def __call__[self] -> Any:
        return self.dispatch_config[self.event_name][]


def successful_event[] -> None:
    print["'successful_event' executed"]


def failed_event[] -> None:
    print["'failed_event' executed"]
    1 / 0


def main[] -> None:
    success_hook = EventHook["success"]
    failure_hook = EventHook["failure"]

    with ExitStack[] as stack:
        try:
            # Run successful event and attach success hook.
            successful_event[]
            stack.callback[success_hook]

            failed_event[]
        except ZeroDivisionError:
            # When the failed even raises an error, attach failure hook.
            stack.callback[failure_hook]


if __name__ == "__main__":
    main[]

0 của Python có thể được sử dụng làm cả trình trang trí và trình quản lý ngữ cảnh. Đối với việc vá và hủy vá chi tiết trong quá trình kiểm tra, cách tiếp cận của trình quản lý ngữ cảnh cho phép bạn kiểm soát nhiều hơn so với đối tác trang trí của nó. Trong trường hợp này,

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

1 có thể giúp bạn tránh được nhiều lồng giống như trong phần trước

# src.py
from __future__ import annotations

from contextlib import ExitStack
from http import HTTPStatus
from typing import Any
from unittest.mock import patch
import httpx


def get[url: str] -> dict[str, Any]:
    return httpx.get[url].json[]


def post[url: str, data: dict[str, Any]] -> dict[str, Any]:
    return httpx.post[url, json=data].json[]


def main[] -> dict[str, Any]:
    res_get = get["//httpbin.org/get"]
    res_post = post["//httpbin.org/post", {"foo": "bar"}]

    return {"get": res_get, "post": res_post}


def test_main[] -> None:
    with ExitStack[] as stack:
        # Arrange
        mock_httpx_get = stack.enter_context[
            patch[
                "httpx.get",
                autospec=True,
                return_value=httpx.Response[
                    json={"fizz": "bazz"}, status_code=HTTPStatus.OK
                ],
            ],
        ]
        mock_httpx_post = stack.enter_context[
            patch[
                "httpx.post",
                autospec=True,
                return_value=httpx.Response[
                    json={"foo": "bar"}, status_code=HTTPStatus.CREATED
                ],
            ]
        ]

        # Act
        res = main[]

        # Assert
        assert res["get"]["fizz"] == "bazz"
        assert res["post"]["foo"] == "bar"
        assert mock_httpx_get.call_count == 1
        assert mock_httpx_post.call_count == 1

Chạy đoạn mã trên với pytest sẽ cho thấy rằng bài kiểm tra đã vượt qua mà không có bất kỳ lỗi nào

src.py::test_main PASSED

======================= 1 passed in 0.11s =======================

Ở đây, tôi đang thực hiện các yêu cầu

# src.py
from __future__ import annotations

from contextlib import ExitStack
from typing import Any


class EventHook:
    def __init__[self, event_name: str] -> None:
        self.event_name = event_name
        self.dispatch_config = {
            "success": self.on_success,
            "failure": self.on_failure,
        }

    def on_success[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def on_failure[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def __call__[self] -> Any:
        return self.dispatch_config[self.event_name][]


def successful_event[] -> None:
    print["'successful_event' executed"]


def failed_event[] -> None:
    print["'failed_event' executed"]
    1 / 0


def main[] -> None:
    success_hook = EventHook["success"]
    failure_hook = EventHook["failure"]

    with ExitStack[] as stack:
        try:
            # Run successful event and attach success hook.
            successful_event[]
            stack.callback[success_hook]

            failed_event[]
        except ZeroDivisionError:
            # When the failed even raises an error, attach failure hook.
            stack.callback[failure_hook]


if __name__ == "__main__":
    main[]

2 và

INFO:root:Request fec8fc9f-7762-4d53-b8f9-3dc7802108a4 completed successfully.
INFO:root:Request 4b6ed0ed-b7cf-46f0-9374-85627be4c26c completed successfully.

0 với thư viện

# src.py
from __future__ import annotations

from contextlib import ExitStack
from typing import Any


class EventHook:
    def __init__[self, event_name: str] -> None:
        self.event_name = event_name
        self.dispatch_config = {
            "success": self.on_success,
            "failure": self.on_failure,
        }

    def on_success[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def on_failure[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def __call__[self] -> Any:
        return self.dispatch_config[self.event_name][]


def successful_event[] -> None:
    print["'successful_event' executed"]


def failed_event[] -> None:
    print["'failed_event' executed"]
    1 / 0


def main[] -> None:
    success_hook = EventHook["success"]
    failure_hook = EventHook["failure"]

    with ExitStack[] as stack:
        try:
            # Run successful event and attach success hook.
            successful_event[]
            stack.callback[success_hook]

            failed_event[]
        except ZeroDivisionError:
            # When the failed even raises an error, attach failure hook.
            stack.callback[failure_hook]


if __name__ == "__main__":
    main[]

4 và trong hàm

# src.py
from __future__ import annotations

from contextlib import ExitStack
from typing import Any


class EventHook:
    def __init__[self, event_name: str] -> None:
        self.event_name = event_name
        self.dispatch_config = {
            "success": self.on_success,
            "failure": self.on_failure,
        }

    def on_success[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def on_failure[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def __call__[self] -> Any:
        return self.dispatch_config[self.event_name][]


def successful_event[] -> None:
    print["'successful_event' executed"]


def failed_event[] -> None:
    print["'failed_event' executed"]
    1 / 0


def main[] -> None:
    success_hook = EventHook["success"]
    failure_hook = EventHook["failure"]

    with ExitStack[] as stack:
        try:
            # Run successful event and attach success hook.
            successful_event[]
            stack.callback[success_hook]

            failed_event[]
        except ZeroDivisionError:
            # When the failed even raises an error, attach failure hook.
            stack.callback[failure_hook]


if __name__ == "__main__":
    main[]

5, khả năng gọi

# src.py
from __future__ import annotations

from contextlib import ExitStack
from typing import Any


class EventHook:
    def __init__[self, event_name: str] -> None:
        self.event_name = event_name
        self.dispatch_config = {
            "success": self.on_success,
            "failure": self.on_failure,
        }

    def on_success[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def on_failure[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def __call__[self] -> Any:
        return self.dispatch_config[self.event_name][]


def successful_event[] -> None:
    print["'successful_event' executed"]


def failed_event[] -> None:
    print["'failed_event' executed"]
    1 / 0


def main[] -> None:
    success_hook = EventHook["success"]
    failure_hook = EventHook["failure"]

    with ExitStack[] as stack:
        try:
            # Run successful event and attach success hook.
            successful_event[]
            stack.callback[success_hook]

            failed_event[]
        except ZeroDivisionError:
            # When the failed even raises an error, attach failure hook.
            stack.callback[failure_hook]


if __name__ == "__main__":
    main[]

6 và

# src.py
from __future__ import annotations

from contextlib import ExitStack
from typing import Any


class EventHook:
    def __init__[self, event_name: str] -> None:
        self.event_name = event_name
        self.dispatch_config = {
            "success": self.on_success,
            "failure": self.on_failure,
        }

    def on_success[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def on_failure[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def __call__[self] -> Any:
        return self.dispatch_config[self.event_name][]


def successful_event[] -> None:
    print["'successful_event' executed"]


def failed_event[] -> None:
    print["'failed_event' executed"]
    1 / 0


def main[] -> None:
    success_hook = EventHook["success"]
    failure_hook = EventHook["failure"]

    with ExitStack[] as stack:
        try:
            # Run successful event and attach success hook.
            successful_event[]
            stack.callback[success_hook]

            failed_event[]
        except ZeroDivisionError:
            # When the failed even raises an error, attach failure hook.
            stack.callback[failure_hook]


if __name__ == "__main__":
    main[]

7 được vá bằng trình quản lý ngữ cảnh

# src.py
from __future__ import annotations

from contextlib import ExitStack
from typing import Any


class EventHook:
    def __init__[self, event_name: str] -> None:
        self.event_name = event_name
        self.dispatch_config = {
            "success": self.on_success,
            "failure": self.on_failure,
        }

    def on_success[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def on_failure[self] -> None:
        print[f"'{self.event_name}' hook called"]

    def __call__[self] -> Any:
        return self.dispatch_config[self.event_name][]


def successful_event[] -> None:
    print["'successful_event' executed"]


def failed_event[] -> None:
    print["'failed_event' executed"]
    1 / 0


def main[] -> None:
    success_hook = EventHook["success"]
    failure_hook = EventHook["failure"]

    with ExitStack[] as stack:
        try:
            # Run successful event and attach success hook.
            successful_event[]
            stack.callback[success_hook]

            failed_event[]
        except ZeroDivisionError:
            # When the failed even raises an error, attach failure hook.
            stack.callback[failure_hook]


if __name__ == "__main__":
    main[]

8. Tuy nhiên,

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

1 cho phép tôi làm điều đó ở đây mà không cần tạo thêm các khối

INFO:root:Rolling back request: 50eb2734-f84c-4013-b5f6-0ccf1aa5d79a
INFO:root:Rolling back request: b326e567-a006-4648-bf04-202397f44e31

5 lồng nhau

Trình quản lý bối cảnh Python là gì?

Trình quản lý ngữ cảnh được dùng để thiết lập và loại bỏ các ngữ cảnh tạm thời, thiết lập và giải quyết các cài đặt tùy chỉnh cũng như thu thập và giải phóng tài nguyên . Hàm open[] để mở tệp là một trong những ví dụ quen thuộc nhất của trình quản lý ngữ cảnh.

Chức năng nhập trong Python là gì?

__enter__ và [__exit__] đều là các phương thức được gọi khi nhập và thoát khỏi phần nội dung của "câu lệnh with" [PEP . câu lệnh with có ý định ẩn điều khiển luồng của mệnh đề try last và làm cho mã trở nên khó hiểu.

3 loại câu lệnh điều kiện * của Python là gì?

Python có 3 câu lệnh điều kiện chính mà bạn nên biết. .

câu lệnh if

câu lệnh if-else

thang if-elif-else

Sự khác biệt giữa trình quản lý ngữ cảnh và trình trang trí trong Python là gì?

trình quản lý bối cảnh là các đối tượng được sử dụng với python với từ khóa. Nó chạy mã khi vào khối và thoát khỏi khối. decorators là những sửa đổi đối với một hàm hoặc định nghĩa lớp. Nó chạy mã thay thế chức năng như nó đang được xác định

Thực thi giao dịch cấp yêu cầu

Gọi móc sự kiện có điều kiện

Tránh cấu trúc bối cảnh lồng nhau

Áp dụng nhiều bản vá làm trình quản lý ngữ cảnh

Trình quản lý bối cảnh Python là gì?

Chức năng __ nhập __ trong Python là gì?

3 loại câu lệnh điều kiện * của Python là gì?

Sự khác biệt giữa trình quản lý ngữ cảnh và trình trang trí trong Python là gì?

Bài Viết Liên Quan

Toplist mới

Bài mới nhất

Chủ Đề

Chức năng nhập trong Python là gì?