Python with 关键字

note = open("zhou-review.txt", "w", encoding="utf-8")

try:
    note.write("ZHOU：今天复习了循环和条件判断。\n")
finally:
    note.close()

print(note.closed)

True

with open("zhou-review.txt", "w", encoding="utf-8") as note:
    note.write("ZHOU：今天复习了循环和条件判断。\n")

print(note.closed)

with expression as target:
    需要在上下文中执行的代码

with open("zhou-plan.txt", "w", encoding="utf-8") as writer:
    writer.write("复习 with\n完成两道练习\n")

with open("zhou-plan.txt", "r", encoding="utf-8") as reader:
    plan = reader.read()
    print("代码块内，文件已关闭：", reader.closed)

print("代码块外，文件已关闭：", reader.closed)
print(plan, end="")

代码块内，文件已关闭： False
代码块外，文件已关闭： True
复习 with
完成两道练习

class StudySession:
    def __init__(self, learner):
        self.learner = learner

    def __enter__(self):
        print(f"开始学习：{self.learner}")
        return {"learner": self.learner, "topic": "with"}

    def __exit__(self, exc_type, exc_value, traceback):
        print("结束学习：整理记录")
        return False


session = StudySession("ZHOU")

with session as info:
    print(f"当前主题：{info['topic']}")
    print("as 绑定的对象是否就是管理器：", info is session)

print("继续下一项任务")

开始学习：ZHOU
当前主题：with
as 绑定的对象是否就是管理器： False
结束学习：整理记录
继续下一项任务

from io import StringIO

buffer = StringIO()

try:
    with buffer as report:
        report.write("ZHOU 的成绩记录\n")
        raise ValueError("成绩必须在 0 到 100 之间")
except ValueError as error:
    print("捕获异常：", error)

print("文本流已关闭：", buffer.closed)

捕获异常： 成绩必须在 0 到 100 之间
文本流已关闭： True

with open("zhou-topics.txt", "w", encoding="utf-8") as seed:
    seed.write("条件判断\n循环语句\n上下文管理\n")

with (
    open("zhou-topics.txt", "r", encoding="utf-8") as source,
    open("zhou-topics-numbered.txt", "w", encoding="utf-8") as target,
):
    for number, topic in enumerate(source, start=1):
        target.write(f"{number}. {topic.strip()}\n")

print(source.closed, target.closed)

with open("zhou-topics-numbered.txt", "r", encoding="utf-8") as result:
    print(result.read(), end="")

True True
1. 条件判断
2. 循环语句
3. 上下文管理

from contextlib import contextmanager


@contextmanager
def temporary_mode(settings, mode):
    previous = settings["mode"]
    settings["mode"] = mode

    try:
        yield settings
    finally:
        settings["mode"] = previous


settings = {"mode": "review"}
print("进入前：", settings["mode"])

try:
    with temporary_mode(settings, "practice") as current:
        print("上下文内：", current["mode"])
        raise RuntimeError("练习被中断")
except RuntimeError as error:
    print("捕获异常：", error)

print("退出后：", settings["mode"])

进入前： review
上下文内： practice
捕获异常： 练习被中断
退出后： review

from threading import Lock, Thread

guard = Lock()
completed = 0


def record_one():
    global completed
    with guard:
        completed += 1


workers = [Thread(target=record_one) for _ in range(6)]

for worker in workers:
    worker.start()

for worker in workers:
    worker.join()

print("已完成练习：", completed)

已完成练习： 6

from decimal import Decimal, getcontext, localcontext

original_precision = getcontext().prec

with localcontext() as context:
    context.prec = 5
    result = Decimal("2") / Decimal("7")
    print("本次计算：", result)

print("精度已恢复：", getcontext().prec == original_precision)
print("计算结果仍可用：", result)

本次计算： 0.28571
精度已恢复： True
计算结果仍可用： 0.28571

import sqlite3
from contextlib import closing

with closing(sqlite3.connect(":memory:", isolation_level="DEFERRED")) as connection:
    connection.execute(
        "CREATE TABLE practice (id INTEGER PRIMARY KEY, topic TEXT NOT NULL)"
    )

    try:
        with connection:
            connection.execute(
                "INSERT INTO practice VALUES (?, ?)",
                (1, "with 关键字"),
            )
            connection.execute(
                "INSERT INTO practice VALUES (?, ?)",
                (1, "上下文管理器"),
            )
    except sqlite3.IntegrityError:
        print("编号重复，本次写入已回滚")

    count = connection.execute("SELECT COUNT(*) FROM practice").fetchone()[0]
    print("当前记录数：", count)

print("已退出外层 closing，连接已关闭")

编号重复，本次写入已回滚
当前记录数： 0
已退出外层 closing，连接已关闭