第 2 周 练习骨架(只读展示)
怎么用
这里只是为了方便在手机/平板上看题。实际编码请在 VS Code 里打开 exercises/week2/ 下的对应文件,把 TODO 换成你的实现,然后运行:
uv run pytest exercises/week2/test_文件名.py -q(目录型练习用 uv run pytest exercises/week2/目录名 -q)
任务规格、测试用例与陷阱反例见对应的计划页。
test_w2_03_typing.py
| exercises/week2/test_w2_03_typing.py |
|---|
| """Week 2 · M7 · w2_03_typing.py 的测试
用法:uv run pytest exercises/week2/test_w2_03_typing.py -q
注意:这里**只测运行时行为**。类型注解写得对不对,pytest 是看不出来的,
所以本练习另有两条硬要求:
1) VS Code 里 Pylance(`python.analysis.typeCheckingMode = "standard"`)零报错;
2) `uvx ty check exercises/week2/w2_03_typing.py` 零错误。
两条都做到,才算这个文件过关。
"""
import pytest
from w2_03_typing import (
LAST_DURATIONS,
MAX_STACK,
ListRepository,
Repository,
Stack,
describe_user,
is_json,
make_user,
open_mode,
pairwise,
parse,
timed,
)
def test_max_stack_is_a_constant() -> None:
"""Final 常量只是给类型检查器的约定,运行时就是个普通 int。"""
assert MAX_STACK == 100
@pytest.mark.parametrize(
"value",
[None, True, 0, 1.5, "s", [], {}, [1, "a", None], {"a": {"b": [1, 2]}}],
)
def test_is_json_true(value: object) -> None:
"""能被 json.dumps 直接序列化的值都是 Json。"""
assert is_json(value) is True
@pytest.mark.parametrize(
"value",
[{"a", "b"}, b"bytes", object(), {1: "a"}, [1, {2: 3}], {"a": object()}],
)
def test_is_json_false(value: object) -> None:
"""set/bytes/对象、以及非 str 键的 dict 都不是 Json。"""
assert is_json(value) is False
def test_stack_push_pop_peek() -> None:
"""push 返回 self,所以能链式调用;pop 后长度减一。"""
s: Stack[int] = Stack()
assert s.is_empty() is True
assert s.push(1).push(2) is s
assert len(s) == 2
assert s.peek() == 2
assert len(s) == 2
assert s.pop() == 2
assert s.pop() == 1
assert s.is_empty() is True
def test_stack_pop_empty_raises() -> None:
"""空栈弹出应抛 IndexError。"""
s: Stack[str] = Stack()
with pytest.raises(IndexError):
s.pop()
with pytest.raises(IndexError):
s.peek()
@pytest.mark.parametrize(
("xs", "expected"),
[
([1, 2, 3], [(1, 2), (2, 3)]),
(["a"], []),
([], []),
((1, 2), [(1, 2)]),
],
)
def test_pairwise(xs: object, expected: object) -> None:
"""相邻配对;元组也是 Sequence,所以也能传。"""
assert pairwise(xs) == expected # type: ignore[arg-type]
def test_timed_keeps_metadata_and_result() -> None:
"""functools.wraps 保留名字和 docstring,返回值原样返回。"""
@timed
def add(a: int, b: int = 0) -> int:
"""加法。"""
return a + b
assert add.__name__ == "add"
assert add.__doc__ == "加法。"
assert add(1, 2) == 3
assert add(1, b=5) == 6
assert LAST_DURATIONS["add"] >= 0
def test_timed_lets_exceptions_through() -> None:
"""装饰器不能吞异常。"""
@timed
def boom() -> None:
raise ValueError("炸了")
with pytest.raises(ValueError, match="炸了"):
boom()
def test_make_user_required_and_optional() -> None:
"""name 去空白;email 没传就不出现在字典里。"""
assert make_user(name=" 小明 ", age=15) == {"name": "小明", "age": 15}
user = make_user(name="a", age=1, email="a@b.com")
assert user["email"] == "a@b.com"
assert "email" not in make_user(name="a", age=1)
def test_make_user_rejects_negative_age() -> None:
"""age 为负要抛 ValueError。"""
with pytest.raises(ValueError, match="age"):
make_user(name="a", age=-1)
def test_describe_user() -> None:
"""NotRequired 的键要用 .get 取。"""
assert describe_user({"name": "小明", "age": 15}) == "小明(15 岁)"
assert (
describe_user({"name": "a", "age": 1, "email": "x@y.z"}) == "a(1 岁,x@y.z)"
)
def test_open_mode() -> None:
"""Literal 的两个合法值。"""
assert open_mode("r") == "只读"
assert open_mode("w") == "只写"
def test_open_mode_rejects_others() -> None:
"""运行时来的脏数据仍要拦(类型检查器管不到)。"""
with pytest.raises(ValueError, match="不支持的模式"):
open_mode("rb") # type: ignore[arg-type]
def test_parse_overloads() -> None:
"""str -> int,bytes -> str。"""
assert parse("42") == 42
assert parse(b"hi") == "hi"
def test_parse_rejects_other_types() -> None:
"""别的类型抛 TypeError。"""
with pytest.raises(TypeError):
parse(1.5) # type: ignore[call-overload]
def test_parse_bad_str_raises_value_error() -> None:
"""int("abc") 自然会抛 ValueError,不用自己转成别的异常。"""
with pytest.raises(ValueError):
parse("abc")
def test_list_repository_satisfies_protocol() -> None:
"""结构化子类型:没有继承 Repository,也是 Repository。"""
repo: Repository[int] = ListRepository[int]()
repo.add(1)
repo.add(2)
assert repo.list_all() == [1, 2]
assert repo.get(0) == 1
assert repo.get(9) is None
assert isinstance(repo, Repository)
def test_list_all_returns_a_copy() -> None:
"""返回副本,外部改动不影响内部。"""
repo = ListRepository[str]()
repo.add("a")
got = repo.list_all()
got.append("b")
assert repo.list_all() == ["a"]
def test_plain_object_is_not_a_repository() -> None:
"""@runtime_checkable 的 isinstance 只看方法名在不在。"""
assert not isinstance(object(), Repository)
|
test_w2_04_pydantic_models.py
| exercises/week2/test_w2_04_pydantic_models.py |
|---|
| """Week 2 · M8a · w2_04_pydantic_models.py 的测试
用法:uv run pytest exercises/week2/test_w2_04_pydantic_models.py -q
读法:每个测试名就是一条需求。先读测试,再去实现。
"""
from datetime import UTC, datetime, timedelta
from decimal import Decimal
import pytest
from pydantic import ValidationError
from w2_04_pydantic_models import (
Address,
Meeting,
Money,
Note,
Profile,
format_errors,
parse_notes,
promote,
)
def test_note_minimal() -> None:
"""只给 title 就能建出来:其他字段都有默认值。"""
note = Note(title="标题")
assert note.title == "标题"
assert note.body == ""
assert note.tags == []
assert note.priority == "normal"
def test_note_created_at_is_aware_utc() -> None:
"""created_at 默认是"现在",而且必须带时区(存库一律 UTC)。"""
note = Note(title="t")
assert note.created_at.tzinfo is not None
assert abs((datetime.now(UTC) - note.created_at).total_seconds()) < 5
def test_note_strips_whitespace() -> None:
"""str_strip_whitespace=True:所有 str 字段自动去首尾空白。"""
assert Note(title=" 标题 ").title == "标题"
def test_note_title_too_short() -> None:
"""去空白后为空 -> min_length=1 失败。"""
with pytest.raises(ValidationError) as exc_info:
Note(title=" ")
assert exc_info.value.errors()[0]["type"] == "string_too_short"
def test_note_title_too_long() -> None:
"""超过 100 字要报错。"""
with pytest.raises(ValidationError):
Note(title="字" * 101)
def test_note_tags_lowercase_and_dedup() -> None:
"""标签小写、去空白、按出现顺序去重。"""
note = Note(title="t", tags=["Python", "python", " AI ", "ai", ""])
assert note.tags == ["python", "ai"]
def test_note_too_many_tags() -> None:
"""标签最多 5 个(这个检查在去重之前)。"""
with pytest.raises(ValidationError) as exc_info:
Note(title="t", tags=["a", "b", "c", "d", "e", "f"])
assert exc_info.value.errors()[0]["type"] == "too_long"
def test_note_lax_mode_coerces_int_like_str() -> None:
"""默认宽松模式:priority 是 Literal,写错的值要被拒绝。"""
with pytest.raises(ValidationError):
Note(title="t", priority="urgent") # type: ignore[arg-type]
def test_note_forbids_extra_fields() -> None:
"""extra="forbid":拼错的键名当场报错,而不是被悄悄忽略。"""
with pytest.raises(ValidationError) as exc_info:
Note(title="t", titel="打错了") # type: ignore[call-arg]
assert exc_info.value.errors()[0]["type"] == "extra_forbidden"
def test_mutable_default_is_not_shared() -> None:
"""陷阱题:Pydantic 会深拷贝默认值,所以两个实例不共享同一个列表。
(和 dataclass 相反——dataclass 里 `tags: list = []` 直接是语法错误。)
"""
a = Note(title="a")
b = Note(title="b")
a.tags.append("x")
assert b.tags == []
def test_address_ok() -> None:
"""6 位邮编通过。"""
addr = Address(city="北京", street="中关村 1 号", zipcode="100080")
assert addr.zipcode == "100080"
@pytest.mark.parametrize("zipcode", ["1008", "1000800", "abcdef", ""])
def test_address_bad_zipcode(zipcode: str) -> None:
"""不是 6 位数字就报错。"""
with pytest.raises(ValidationError):
Address(city="北京", zipcode=zipcode)
def test_address_city_required_non_empty() -> None:
"""city 至少 1 个字。"""
with pytest.raises(ValidationError):
Address(city="", zipcode="100080")
def test_profile_nested_dict_becomes_model() -> None:
"""嵌套模型:传 dict 会被自动解析成 Address 实例。"""
profile = Profile(
name="小明", age=15, address={"city": "北京", "zipcode": "100080"}
)
assert isinstance(profile.address, Address)
assert profile.address.city == "北京"
assert profile.nickname is None
@pytest.mark.parametrize("age", [-1, 151])
def test_profile_age_out_of_range(age: int) -> None:
"""age 必须在 0–150。"""
with pytest.raises(ValidationError):
Profile(name="小明", age=age, address={"city": "北京", "zipcode": "100080"})
def test_meeting_ok() -> None:
"""start < end 正常。"""
start = datetime(2026, 9, 14, 9, 0, tzinfo=UTC)
meeting = Meeting(title="周会", start=start, end=start + timedelta(hours=1))
assert meeting.end > meeting.start
def test_meeting_bad_order() -> None:
"""start >= end 时 model_validator 要拦住。"""
start = datetime(2026, 9, 14, 9, 0, tzinfo=UTC)
with pytest.raises(ValidationError) as exc_info:
Meeting(title="周会", start=start, end=start)
assert "start 必须早于 end" in str(exc_info.value)
def test_meeting_parses_iso_string() -> None:
"""datetime 字段接受 ISO 8601 字符串(这是 Pydantic 的解析能力)。"""
meeting = Meeting(
title="周会",
start="2026-09-14T09:00:00+00:00", # type: ignore[arg-type]
end="2026-09-14T10:00:00+00:00", # type: ignore[arg-type]
)
assert meeting.start.hour == 9
def test_money_ok() -> None:
"""严格模式下 Decimal 只接受 Decimal。"""
money = Money(amount=Decimal("1.5"), currency="CNY")
assert money.amount == Decimal("1.5")
def test_money_strict_rejects_str_amount() -> None:
"""strict=True:字符串 "1.5" 不再被悄悄转成 Decimal。"""
with pytest.raises(ValidationError):
Money(amount="1.5", currency="CNY") # type: ignore[arg-type]
def test_money_rejects_bad_currency() -> None:
"""currency 必须是 3 位大写字母。"""
with pytest.raises(ValidationError):
Money(amount=Decimal("1"), currency="cny")
def test_money_rejects_non_positive_amount() -> None:
"""amount 必须大于 0。"""
with pytest.raises(ValidationError):
Money(amount=Decimal("0"), currency="CNY")
def test_format_errors() -> None:
"""errors() 的 loc 用 "." 连起来,前面带上第几条。"""
with pytest.raises(ValidationError) as exc_info:
Note(title="")
messages = format_errors(exc_info.value, 0)
assert len(messages) == 1
assert messages[0].startswith("第 0 条 title: ")
def test_format_errors_nested_loc() -> None:
"""嵌套字段的 loc 是多段,比如 address.zipcode。"""
with pytest.raises(ValidationError) as exc_info:
Profile(name="小明", age=1, address={"city": "北京", "zipcode": "x"})
messages = format_errors(exc_info.value, 3)
assert any("address.zipcode" in m for m in messages)
assert all(m.startswith("第 3 条 ") for m in messages)
def test_parse_notes_partial_success() -> None:
"""一条坏数据不能拖垮整批。"""
notes, errors = parse_notes(
[
{"title": "好的"},
{"title": ""},
{"title": "也好的", "tags": ["A"]},
{"title": "多字段", "oops": 1},
]
)
assert [n.title for n in notes] == ["好的", "也好的"]
assert len(errors) == 2
assert errors[0].startswith("第 1 条 ")
assert errors[1].startswith("第 3 条 ")
def test_parse_notes_empty() -> None:
"""空输入返回两个空列表。"""
assert parse_notes([]) == ([], [])
def test_promote_returns_copy() -> None:
"""model_copy(update=) 返回副本,原对象不变。"""
note = Note(title="t")
high = promote(note)
assert high.priority == "high"
assert note.priority == "normal"
assert high is not note
assert high.title == note.title
|
test_w2_05_pydantic_schema_settings.py
| exercises/week2/test_w2_05_pydantic_schema_settings.py |
|---|
| """Week 2 · M8b · w2_05_pydantic_schema_settings.py 的测试
用法:uv run pytest exercises/week2/test_w2_05_pydantic_schema_settings.py -q
要点:Settings 的测试用 monkeypatch.setenv 造环境变量、monkeypatch.chdir 换工作目录,
这样就不会依赖你机器上真实存在的 .env——测试必须可重复。
"""
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any
import pytest
from pydantic import ValidationError
from w2_05_pydantic_schema_settings import (
ApiNote,
Article,
Click,
Doc,
KeyPress,
Settings,
describe_settings,
dump_for_api,
parse_event,
parse_events,
schema_summary,
time_validation,
validate_docs,
validate_docs_one_by_one,
)
DOCS: list[dict[str, Any]] = [
{"id": 1, "title": "a", "created_at": "2026-09-14T00:00:00Z"},
{"id": 2, "title": "b", "created_at": "2026-09-15T00:00:00Z"},
]
def test_parse_event_click() -> None:
"""kind="click" 解析成 Click。"""
event = parse_event({"kind": "click", "x": 1, "y": 2})
assert isinstance(event, Click)
assert (event.x, event.y) == (1, 2)
def test_parse_event_key() -> None:
"""kind="key" 解析成 KeyPress。"""
event = parse_event({"kind": "key", "key": "a"})
assert isinstance(event, KeyPress)
assert event.key == "a"
def test_parse_event_unknown_kind() -> None:
"""判别字段不认识时,报错会直接指向 kind。"""
with pytest.raises(ValidationError) as exc_info:
parse_event({"kind": "scroll"})
assert exc_info.value.errors()[0]["type"] == "union_tag_invalid"
def test_parse_event_missing_kind() -> None:
"""kind 缺失也要报错(这就是判别联合比"两个都试一遍"好的地方)。"""
with pytest.raises(ValidationError):
parse_event({"x": 1, "y": 2})
def test_parse_events_mixed() -> None:
"""一批混合事件。"""
events = parse_events(
[{"kind": "click", "x": 0, "y": 0}, {"kind": "key", "key": "b"}]
)
assert [type(e).__name__ for e in events] == ["Click", "KeyPress"]
def test_dump_for_api_json_mode() -> None:
"""mode="json":datetime -> ISO 字符串,Decimal -> 字符串。"""
note = ApiNote(
title="t", createdAt=datetime(2026, 9, 14, tzinfo=UTC), price=Decimal("1.5")
)
dumped = dump_for_api(note)
assert dumped["createdAt"] == "2026-09-14T00:00:00Z"
assert dumped["price"] == "1.5"
def test_dump_for_api_excludes_none_and_uses_alias() -> None:
"""exclude_none=True 丢掉 None 字段;by_alias=True 用驼峰键名。"""
note = ApiNote(
title="t", created_at=datetime(2026, 9, 14, tzinfo=UTC), price=Decimal("2")
)
dumped = dump_for_api(note)
assert "note_id" not in dumped
assert "created_at" not in dumped
assert dumped["summary"] == "t(2 元)"
def test_dump_for_api_result_is_json_serializable() -> None:
"""mode="json" 的意义就是结果能直接喂给 json.dumps。"""
import json
note = ApiNote(
title="t", createdAt=datetime(2026, 9, 14, tzinfo=UTC), price=Decimal("1")
)
assert json.dumps(dump_for_api(note), ensure_ascii=False).startswith("{")
def test_schema_summary_required() -> None:
"""只有没有默认值的字段是必填。"""
assert schema_summary(Article)["required"] == ["title"]
def test_schema_summary_simple_field() -> None:
"""description 和 default 都来自 Field(...)。"""
fields = schema_summary(Article)["fields"]
assert fields["title"] == {
"type": "string",
"description": "文章标题",
"default": None,
}
assert fields["views"]["type"] == "integer"
assert fields["views"]["default"] == 0
def test_schema_summary_optional_field_uses_any_of() -> None:
"""str | None 在 JSON Schema 里是 anyOf。"""
fields = schema_summary(Article)["fields"]
assert fields["author"]["type"] == "string|null"
assert fields["author"]["description"] == "作者,可省略"
def test_schema_summary_list_field() -> None:
"""default_factory 的字段在 schema 里没有 default,所以取到 None。"""
fields = schema_summary(Article)["fields"]
assert fields["tags"]["type"] == "array"
assert fields["tags"]["default"] is None
def test_schema_summary_covers_all_fields() -> None:
"""摘要要把每个字段都列出来。"""
assert set(schema_summary(Article)["fields"]) == {
"title",
"views",
"author",
"tags",
}
def test_settings_defaults(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""没有环境变量、没有 .env 时用默认值。"""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("KB_DB_PATH", raising=False)
monkeypatch.delenv("KB_LOG_LEVEL", raising=False)
settings = Settings()
assert settings.db_path == Path("kb.db")
assert settings.log_level == "INFO"
def test_settings_from_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""env_prefix="KB_":db_path 读的是 KB_DB_PATH,并自动转成 Path。"""
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("KB_DB_PATH", "data/kb.sqlite3")
monkeypatch.setenv("KB_LOG_LEVEL", "DEBUG")
settings = Settings()
assert settings.db_path == Path("data/kb.sqlite3")
assert settings.log_level == "DEBUG"
def test_settings_env_is_case_insensitive(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""环境变量名大小写不敏感。"""
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("kb_log_level", "ERROR")
assert Settings().log_level == "ERROR"
def test_settings_reads_dotenv(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
""".env 也能读到(真实项目里 .env 不进 Git,仓库只放 .env.example)。"""
(tmp_path / ".env").write_text("KB_LOG_LEVEL=WARNING\n", encoding="utf-8")
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("KB_LOG_LEVEL", raising=False)
assert Settings().log_level == "WARNING"
def test_settings_env_beats_dotenv(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""优先级:环境变量 > .env > 默认值。"""
(tmp_path / ".env").write_text("KB_LOG_LEVEL=WARNING\n", encoding="utf-8")
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("KB_LOG_LEVEL", "DEBUG")
assert Settings().log_level == "DEBUG"
def test_settings_rejects_bad_level(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Literal 会拦住写错的级别(比如小写的 info)。"""
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("KB_LOG_LEVEL", "info")
with pytest.raises(ValidationError):
Settings()
def test_describe_settings(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""一句话描述配置。"""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("KB_DB_PATH", raising=False)
monkeypatch.delenv("KB_LOG_LEVEL", raising=False)
assert describe_settings(Settings()) == "db=kb.db level=INFO"
def test_validate_docs() -> None:
"""TypeAdapter 不用建模型也能校验 list[Doc]。"""
docs = validate_docs(DOCS)
assert [d.id for d in docs] == [1, 2]
assert all(isinstance(d, Doc) for d in docs)
assert docs[0].created_at == datetime(2026, 9, 14, tzinfo=UTC)
def test_validate_docs_same_as_loop() -> None:
"""两种方式结果必须一致(Pydantic 模型之间可以直接 == 比较)。"""
assert validate_docs(DOCS) == validate_docs_one_by_one(DOCS)
def test_validate_docs_error_loc_has_index() -> None:
"""整批校验时,错误的 loc 第一段是出错的下标。"""
bad = [DOCS[0], {"id": "x", "title": "b", "created_at": "2026-09-15T00:00:00Z"}]
with pytest.raises(ValidationError) as exc_info:
validate_docs(bad)
assert exc_info.value.errors()[0]["loc"][0] == 1
def test_time_validation_returns_two_durations() -> None:
"""返回两个非负的秒数。"""
adapter_seconds, loop_seconds = time_validation(DOCS * 10, repeat=2)
assert adapter_seconds >= 0
assert loop_seconds >= 0
def test_time_validation_rejects_bad_repeat() -> None:
"""repeat 必须为正。"""
with pytest.raises(ValueError, match="repeat"):
time_validation(DOCS, repeat=0)
|
test_w2_06_logging_config.py
| exercises/week2/test_w2_06_logging_config.py |
|---|
| """Week 2 · M9 · w2_06_logging_config.py 的测试
用法:uv run pytest exercises/week2/test_w2_06_logging_config.py -q
要点:
- caplog 抓日志(它自己往根 logger 挂了 handler,所以 setup_logging 不能乱清 handler)
- tmp_path 给临时目录写日志文件与 TOML
- 每个测试跑完都把根 logger 恢复原样,否则测试之间会互相污染
"""
import json
import logging
import tomllib
from collections.abc import Iterator
from pathlib import Path
import pytest
from pydantic import ValidationError
from w2_06_logging_config import (
AppConfig,
JsonFormatter,
divide,
load_config,
managed_handlers,
resolve_setting,
setup_logging,
)
@pytest.fixture(autouse=True)
def _isolate_root_logger() -> Iterator[None]:
"""每个测试前后把根 logger 的 handler 和级别恢复原样。"""
root = logging.getLogger()
old_handlers = root.handlers[:]
old_level = root.level
yield
for handler in root.handlers[:]:
if handler not in old_handlers:
root.removeHandler(handler)
handler.close()
root.handlers[:] = old_handlers
root.setLevel(old_level)
def _make_record(
message: str = "a=%s", args: tuple[object, ...] = (1,)
) -> logging.LogRecord:
"""手工造一条 LogRecord,用来单测 Formatter。"""
return logging.LogRecord(
name="demo",
level=logging.INFO,
pathname=__file__,
lineno=1,
msg=message,
args=args,
exc_info=None,
)
def test_json_formatter_keys() -> None:
"""四个固定的键,且 message 是套过参数的结果。"""
line = JsonFormatter().format(_make_record())
data = json.loads(line)
assert set(data) == {"time", "level", "logger", "message"}
assert data["level"] == "INFO"
assert data["logger"] == "demo"
assert data["message"] == "a=1"
def test_json_formatter_keeps_chinese() -> None:
"""ensure_ascii=False:中文不能变成 \\uXXXX。"""
line = JsonFormatter().format(_make_record("你好 %s", ("小明",)))
assert "你好 小明" in line
assert json.loads(line)["message"] == "你好 小明"
def test_json_formatter_with_exception() -> None:
"""有异常信息时多一个 exc_info 键,里面带堆栈。"""
try:
raise ValueError("炸了")
except ValueError:
import sys
record = _make_record("出错了", ())
record.exc_info = sys.exc_info()
data = json.loads(JsonFormatter().format(record))
assert "exc_info" in data
assert "ValueError" in data["exc_info"]
assert "Traceback" in data["exc_info"]
def test_setup_logging_returns_root_and_sets_level() -> None:
"""返回根 logger,级别按参数设置(大小写不敏感)。"""
root = setup_logging("debug")
assert root is logging.getLogger()
assert root.level == logging.DEBUG
def test_setup_logging_rejects_unknown_level() -> None:
"""不认识的级别要报错,而不是悄悄用 INFO。"""
with pytest.raises(ValueError, match="未知日志级别"):
setup_logging("VERBOSE")
def test_setup_logging_adds_one_console_handler() -> None:
"""不给 log_file 时只加一个控制台 handler。"""
setup_logging("INFO")
assert len(managed_handlers()) == 1
def test_setup_logging_is_idempotent(tmp_path: Path) -> None:
"""重复调用不重复加 handler——否则一条日志会打两遍。"""
log_file = tmp_path / "logs" / "app.log"
setup_logging("INFO", log_file)
first = len(managed_handlers())
setup_logging("INFO", log_file)
setup_logging("DEBUG", log_file)
assert first == 2
assert len(managed_handlers()) == 2
def test_setup_logging_keeps_other_handlers(caplog: pytest.LogCaptureFixture) -> None:
"""只摘自己加的 handler:caplog 的 handler 必须活着。"""
with caplog.at_level(logging.INFO):
setup_logging("INFO")
logging.getLogger("demo").info("还能抓到吗")
assert "还能抓到吗" in caplog.text
def test_setup_logging_creates_parent_dir(tmp_path: Path) -> None:
"""父目录不存在也要能写。"""
log_file = tmp_path / "deep" / "nested" / "app.log"
setup_logging("INFO", log_file)
logging.getLogger("demo").info("hi")
assert log_file.exists()
def test_setup_logging_json_file(tmp_path: Path) -> None:
"""json=True 时文件里每行都是一条 JSON。"""
log_file = tmp_path / "app.log"
setup_logging("INFO", log_file, json=True)
logging.getLogger("demo").warning("磁盘快满了 %d%%", 91)
for handler in managed_handlers():
handler.flush()
lines = log_file.read_text(encoding="utf-8").splitlines()
data = json.loads(lines[-1])
assert data["level"] == "WARNING"
assert data["message"] == "磁盘快满了 91%"
def test_setup_logging_plain_file(tmp_path: Path) -> None:
"""json=False 时是普通文本格式,至少能看到级别和消息。"""
log_file = tmp_path / "app.log"
setup_logging("INFO", log_file, json=False)
logging.getLogger("demo").error("坏了")
for handler in managed_handlers():
handler.flush()
text = log_file.read_text(encoding="utf-8")
assert "ERROR" in text
assert "坏了" in text
def test_divide_logs_info(caplog: pytest.LogCaptureFixture) -> None:
"""成功时打一条 INFO,参数用 %s 传(record.args 里能看到)。"""
with caplog.at_level(logging.INFO, logger="w2_06_logging_config"):
assert divide(6, 3) == 2.0
assert len(caplog.records) == 1
record = caplog.records[0]
assert record.levelno == logging.INFO
assert record.args == (6, 3, 2.0)
assert "divide 6 / 3 = 2.0" in caplog.text
def test_divide_by_zero_logs_exception(caplog: pytest.LogCaptureFixture) -> None:
"""除零时返回 None,并且日志里带堆栈(logger.exception 而不是 error)。"""
with caplog.at_level(logging.INFO, logger="w2_06_logging_config"):
assert divide(1, 0) is None
record = caplog.records[0]
assert record.levelno == logging.ERROR
assert record.exc_info is not None
assert "ZeroDivisionError" in caplog.text
assert "Traceback" in caplog.text
def test_load_config_defaults(tmp_path: Path) -> None:
"""只给 name,其余用默认值。"""
path = tmp_path / "config.toml"
path.write_text('name = "kb"\n', encoding="utf-8")
config = load_config(path)
assert config == AppConfig(name="kb")
assert config.max_items == 100
def test_load_config_full(tmp_path: Path) -> None:
"""TOML 里的类型会被原样读出来(数字就是 int,不是字符串)。"""
path = tmp_path / "config.toml"
path.write_text('name = "kb"\nlevel = "DEBUG"\nmax_items = 5\n', encoding="utf-8")
config = load_config(path)
assert (config.level, config.max_items) == ("DEBUG", 5)
def test_load_config_missing_file(tmp_path: Path) -> None:
"""文件不存在就让 FileNotFoundError 抛出去。"""
with pytest.raises(FileNotFoundError):
load_config(tmp_path / "nope.toml")
def test_load_config_bad_toml(tmp_path: Path) -> None:
"""TOML 语法错误由 tomllib 报。"""
path = tmp_path / "config.toml"
path.write_text("name = \n", encoding="utf-8")
with pytest.raises(tomllib.TOMLDecodeError):
load_config(path)
def test_load_config_invalid_value(tmp_path: Path) -> None:
"""字段不合法由 Pydantic 报。"""
path = tmp_path / "config.toml"
path.write_text('name = "kb"\nmax_items = 0\n', encoding="utf-8")
with pytest.raises(ValidationError):
load_config(path)
def test_load_config_unknown_key(tmp_path: Path) -> None:
"""extra="forbid":配置文件里写错的键名要报错。"""
path = tmp_path / "config.toml"
path.write_text('name = "kb"\nlevle = "DEBUG"\n', encoding="utf-8")
with pytest.raises(ValidationError):
load_config(path)
def test_resolve_setting_cli_wins(monkeypatch: pytest.MonkeyPatch) -> None:
"""命令行参数优先级最高。"""
monkeypatch.setenv("KB_LOG_LEVEL", "WARNING")
assert resolve_setting("DEBUG", "KB_LOG_LEVEL", "INFO") == "DEBUG"
def test_resolve_setting_env_then_default(monkeypatch: pytest.MonkeyPatch) -> None:
"""没有命令行参数时看环境变量,再没有就用默认值。"""
monkeypatch.setenv("KB_LOG_LEVEL", "WARNING")
assert resolve_setting(None, "KB_LOG_LEVEL", "INFO") == "WARNING"
monkeypatch.delenv("KB_LOG_LEVEL")
assert resolve_setting(None, "KB_LOG_LEVEL", "INFO") == "INFO"
def test_resolve_setting_empty_env_is_unset(monkeypatch: pytest.MonkeyPatch) -> None:
"""环境变量是空串等于没设。"""
monkeypatch.setenv("KB_LOG_LEVEL", "")
assert resolve_setting(None, "KB_LOG_LEVEL", "INFO") == "INFO"
def test_resolve_setting_empty_cli_counts(monkeypatch: pytest.MonkeyPatch) -> None:
"""命令行显式给空串,就用空串(None 才表示"没给")。"""
monkeypatch.setenv("KB_LOG_LEVEL", "WARNING")
assert resolve_setting("", "KB_LOG_LEVEL", "INFO") == ""
|
test_w2_07_regex.py
| exercises/week2/test_w2_07_regex.py |
|---|
| """Week 2 · M10 · w2_07_regex.py 的测试
用法:uv run pytest exercises/week2/test_w2_07_regex.py -q
"""
import pytest
from w2_07_regex import (
count_words,
extract_cn_phones,
extract_dates,
extract_emails,
mask_secrets,
normalize_whitespace,
parse_log_line,
parse_url,
strip_markdown,
tokenize_zh_en,
)
MARKDOWN = """
# 标题
这是**加粗**的文字,还有 [链接](https://example.com)。
```python
print("hi")
|
小标题
"""
def test_parse_log_line_ok() -> None:
"""五个命名分组都要对;message 里可以再有冒号。"""
line = "2026-09-05 12:00:01 [ERROR] kb.cli: 打不开数据库: 权限不足"
assert parse_log_line(line) == {
"date": "2026-09-05",
"time": "12:00:01",
"level": "ERROR",
"module": "kb.cli",
"message": "打不开数据库: 权限不足",
}
def test_parse_log_line_info() -> None:
"""别的级别也要能解析。"""
parsed = parse_log_line("2026-09-20 08:30:00 [INFO] kb.service: 导入 3 条")
assert parsed is not None
assert parsed["level"] == "INFO"
assert parsed["module"] == "kb.service"
@pytest.mark.parametrize(
"line",
[
"随便一句话",
"2026-9-5 12:00:01 [INFO] a: b",
"2026-09-05 12:00:01 (INFO) a: b",
"2026-09-05 12:00:01 [TRACE] a: b",
"前缀 2026-09-05 12:00:01 [INFO] a: b",
],
)
def test_parse_log_line_none(line: str) -> None:
"""格式不对就返回 None(fullmatch:多一个前缀也不行)。"""
assert parse_log_line(line) is None
def test_extract_emails() -> None:
"""按出现顺序返回。"""
text = "写给 a@b.com 和 x.y+z@mail.co.jp,别写 bad@@x"
assert extract_emails(text) == ["a@b.com", "x.y+z@mail.co.jp"]
def test_extract_emails_none() -> None:
"""没有就返回空列表。"""
assert extract_emails("没有邮箱 @ 这里") == []
def test_extract_cn_phones() -> None:
"""两个合法号码。"""
assert extract_cn_phones("打 13812345678 或 15900001111") == [
"13812345678",
"15900001111",
]
@pytest.mark.parametrize(
"text", ["012345678901234", "12812345678", "1381234567", "138123456789"]
)
def test_extract_cn_phones_rejects(text: str) -> None:
"""第二位不对、位数不对、前后挨着数字,都不算。"""
assert extract_cn_phones(text) == []
def test_extract_dates() -> None:
"""合法日期。"""
assert extract_dates("从 2026-09-14 到 2026-09-20") == [
"2026-09-14",
"2026-09-20",
]
@pytest.mark.parametrize("text", ["2026-13-01", "2026-00-10", "2026-09-32", "20260914"])
def test_extract_dates_rejects(text: str) -> None:
"""月份和日子要在范围内。"""
assert extract_dates(text) == []
def test_normalize_whitespace() -> None:
"""所有空白都算空白。"""
assert normalize_whitespace(" a\tb\n\nc ") == "a b c"
assert normalize_whitespace("") == ""
def test_strip_markdown_document() -> None:
"""代码块整体没了,标题、加粗、链接都变成纯文字。"""
assert strip_markdown(MARKDOWN) == "标题\n\n这是加粗的文字,还有 链接。\n\n小标题"
def test_strip_markdown_inline() -> None:
"""一行的情况。"""
assert strip_markdown("# 标题\n\n粗和链接") == "标题\n\n粗和链接"
def test_strip_markdown_keeps_plain_text() -> None:
"""没有标记的文本原样返回(只 strip)。"""
assert strip_markdown(" 普通文字 ") == "普通文字"
def test_tokenize_zh_en() -> None:
"""中文按字,英文按词,小数不拆。"""
assert tokenize_zh_en("我爱 Python 3.14!") == ["我", "爱", "Python", "3.14"]
def test_tokenize_zh_en_mixed() -> None:
"""英文词和中文字混排。"""
assert tokenize_zh_en("kb 是知识库") == ["kb", "是", "知", "识", "库"]
def test_tokenize_zh_en_drops_punctuation() -> None:
"""标点和空白都不要。"""
assert tokenize_zh_en(",。! \n") == []
def test_mask_secrets() -> None:
"""密钥整段遮掉,手机号只遮中间 4 位。"""
assert (
mask_secrets("key=sk-abcd1234efgh 手机 13812345678")
== "key=sk-* 手机 138**5678"
)
def test_mask_secrets_keeps_short_sk() -> None:
"""sk- 后面不足 8 位的不当密钥处理(避免误伤)。"""
assert mask_secrets("sk-abc") == "sk-abc"
def test_mask_secrets_multiple() -> None:
"""多个都要遮。"""
masked = mask_secrets("13812345678 和 15900001111")
assert masked == "138*5678 和 159*1111"
def test_parse_url_full() -> None:
"""有路径和 query。"""
assert parse_url("https://cnb.cool/me/repo?tab=1") == {
"scheme": "https",
"host": "cnb.cool",
"path": "/me/repo",
}
def test_parse_url_no_path() -> None:
"""没有路径时 path 是空串;端口不进结果。"""
assert parse_url("http://localhost:8000") == {
"scheme": "http",
"host": "localhost",
"path": "",
}
def test_parse_url_fragment() -> None:
"""片段也不进结果。"""
parsed = parse_url("https://example.com/a/b#section")
assert parsed is not None
assert parsed["path"] == "/a/b"
@pytest.mark.parametrize("url", ["ftp://a.b/c", "example.com", "https:/a.b", ""])
def test_parse_url_none(url: str) -> None:
"""协议不对或整串不匹配就返回 None。"""
assert parse_url(url) is None
def test_count_words() -> None:
"""忽略大小写;次数降序,同次数按字母升序。"""
assert count_words("a b A c b") == {"a": 2, "b": 2, "c": 1}
def test_count_words_ignores_non_letters() -> None:
"""数字和中文不算英文单词。"""
assert count_words("hi 你好 123 hi!") == {"hi": 2}
def test_count_words_empty() -> None:
"""空文本返回空字典。"""
assert count_words("") == {}
## `test_w2_08_files_sqlite.py`
```python title="exercises/week2/test_w2_08_files_sqlite.py" linenums="1"
"""w2_08 的测试。运行:uv run pytest exercises/week2/test_w2_08_files_sqlite.py -q"""
import json
import sqlite3
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
import pytest
from w2_08_files_sqlite import (
SqliteNoteRepository,
export_json,
file_sha256,
iter_markdown_files,
json_default,
migrate,
read_csv_with_bom,
utc_now_iso,
)
def test_iter_markdown_files_sorted_and_filtered(tmp_path: Path):
(tmp_path / "b.md").write_text("b", encoding="utf-8")
(tmp_path / "a.MD").write_text("a", encoding="utf-8")
(tmp_path / "c.txt").write_text("c", encoding="utf-8")
(tmp_path / "sub").mkdir()
(tmp_path / "sub" / "d.md").write_text("d", encoding="utf-8")
(tmp_path / ".hidden").mkdir()
(tmp_path / ".hidden" / "e.md").write_text("e", encoding="utf-8")
names = [p.name for p in iter_markdown_files(tmp_path)]
assert names == ["a.MD", "b.md", "d.md"]
def test_iter_markdown_files_missing_root(tmp_path: Path):
with pytest.raises(FileNotFoundError):
list(iter_markdown_files(tmp_path / "nope"))
def test_file_sha256(tmp_path: Path):
p = tmp_path / "x.bin"
p.write_bytes(b"hello")
assert file_sha256(p) == (
"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
)
assert file_sha256(p, chunk_size=2) == file_sha256(p)
def test_json_default():
assert json_default(Decimal("1.5")) == "1.5"
assert json_default({3, 1, 2}) == [1, 2, 3]
assert json_default(datetime(2026, 9, 5, tzinfo=UTC)).startswith(
"2026-09-05T00:00:00"
)
with pytest.raises(TypeError):
json_default(object())
def test_export_json_is_utf8_and_atomic(tmp_path: Path):
target = tmp_path / "out" / "notes.json"
export_json(
target, [{"title": "中文", "n": Decimal("2.5"), "when": datetime(2026, 1, 1)}]
)
text = target.read_text(encoding="utf-8")
assert "中文" in text and "\\u4e2d" not in text
assert json.loads(text)[0]["n"] == "2.5"
assert not list(tmp_path.glob("out/*.tmp"))
def test_export_json_failure_leaves_no_tmp(tmp_path: Path):
target = tmp_path / "bad.json"
with pytest.raises(TypeError):
export_json(target, [{"x": object()}])
assert not target.exists() and not list(tmp_path.glob("*.tmp"))
def test_read_csv_with_bom(tmp_path: Path):
p = tmp_path / "scores.csv"
p.write_bytes("\ufeffname,score\r\n小明,90\r\n".encode())
rows = read_csv_with_bom(p)
assert rows == [{"name": "小明", "score": "90"}]
assert "name" in rows[0] # 没有把 BOM 带进表头
def test_utc_now_iso_has_timezone():
value = utc_now_iso()
parsed = datetime.fromisoformat(value)
assert parsed.tzinfo is not None and parsed.utcoffset().total_seconds() == 0
def test_migrate_is_idempotent(tmp_path: Path):
conn = sqlite3.connect(tmp_path / "m.db")
migrate(conn)
migrate(conn)
tables = {
r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
}
assert "notes" in tables
conn.close()
@pytest.fixture
def repo(tmp_path: Path):
r = SqliteNoteRepository(tmp_path / "data" / "kb.db")
yield r
r.close()
def test_repository_add_get_list(repo: SqliteNoteRepository):
first = repo.add("Python 笔记", "正文", tags=["python", "notes"])
second = repo.add("另一条")
assert (first, second) == (1, 2)
note = repo.get(first)
assert note is not None
assert note["title"] == "Python 笔记" and note["tags"] == ["python", "notes"]
assert repo.get(999) is None
assert [n["id"] for n in repo.list()] == [1, 2]
assert repo.list()[1]["tags"] == []
def test_repository_search_like_and_regex(repo: SqliteNoteRepository):
repo.add("正则表达式入门", "re 模块")
repo.add("Pydantic", "数据校验")
assert [n["title"] for n in repo.search("正则")] == ["正则表达式入门"]
assert [n["title"] for n in repo.search(r"^py", regex=True)] == ["Pydantic"]
assert repo.search("不存在") == []
def test_repository_delete(repo: SqliteNoteRepository):
note_id = repo.add("要删的")
assert repo.delete(note_id) is True
assert repo.delete(note_id) is False
assert repo.list() == []
def test_repository_uses_parameterized_queries(repo: SqliteNoteRepository):
dangerous = "x'; DROP TABLE notes; --"
repo.add(dangerous)
assert repo.search(dangerous)[0]["title"] == dangerous
assert repo.list() # 表还在
w2_01_inline_script.py
| exercises/week2/w2_01_inline_script.py |
|---|
| # /// script
# requires-python = ">=3.14"
# dependencies = ["rich>=14.0"]
# ///
"""Week 2 · M6a · PEP 723 单文件脚本(内联依赖 + rich 表格)
学习目标:知道单文件脚本也能声明自己的依赖——顶部 `# /// script` 块里的
`dependencies` 由 `uv run` 读取,自动准备一个临时环境,不污染项目。
用法:uv run exercises/week2/w2_01_inline_script.py
(本仓库的 .venv 里已经有 rich,所以直接
`uv run python exercises/week2/w2_01_inline_script.py` 也能跑)
做法:这个文件是**范例**,不需要写 TODO,也没有测试。读懂它,然后:
1) 把 `dependencies` 里的 rich 删掉,跑一次看报错,再改回来;
2) 给表格加一列"是否已掌握",值自己填;
3) 用 `uv run --with rich python -c "..."` 达到同样效果,对比两种写法。
预测练习(先猜,再运行):
>>> from rich.table import Table
>>> t = Table(title="x")
>>> t.add_column("a")
>>> t.add_row("1")
>>> print(type(t).__name__, t.row_count)
"""
from rich.console import Console
from rich.table import Table
TOPICS: list[tuple[str, str, str]] = [
("M6", "打包与 Git", "src 布局 / pyproject / uv / Conventional Commits"),
("M7", "typing 进阶", "Protocol / TypedDict / PEP 695 泛型 / TypeIs"),
("M8", "Pydantic", "模型校验 / JSON Schema / Settings"),
("M9", "logging 与配置", "dictConfig / JSON 日志 / 优先级"),
("M10", "正则", "命名分组 / VERBOSE / sub 传函数"),
("M11", "文件与 SQLite", "pathlib / 原子写 / 参数化 SQL"),
]
def build_table(rows: list[tuple[str, str, str]]) -> Table:
"""把 (模块, 主题, 要点) 三元组列表变成一个 rich 表格。
例:build_table([("M6", "打包", "src 布局")]).row_count -> 1
"""
table = Table(title="第 2 周知识地图")
table.add_column("模块", style="cyan", no_wrap=True)
table.add_column("主题", style="bold")
table.add_column("要点")
for module, topic, points in rows:
table.add_row(module, topic, points)
return table
def main() -> None:
"""打印表格——脚本入口。"""
console = Console()
console.print(build_table(TOPICS))
console.print("[green]提示[/green]:单文件脚本适合一次性小工具;要写测试就该建包。")
if __name__ == "__main__":
main()
|
w2_03_typing.py
| exercises/week2/w2_03_typing.py |
|---|
| """Week 2 · M7 · typing 进阶(泛型、TypeIs、ParamSpec、TypedDict、overload)
学习目标:会用 PEP 695 泛型语法(`class Stack[T]`、`def pairwise[T]`);
会写收窄类型的 `TypeIs` 函数、保留签名的装饰器(`**P`/`R`)、
带 `NotRequired` 的 `TypedDict` 与 `Unpack`、`Literal`、`@overload`、泛型
`Protocol`。
用法:uv run pytest exercises/week2/test_w2_03_typing.py -q
另外**必须**:VS Code 里 Pylance(standard 模式)零报错;
`uvx ty check exercises/week2/w2_03_typing.py` 零错误。
做法:把每个 TODO 换成你的实现;注解已经写好,不要改签名。
注解本身就是需求说明——先读注解,再读 docstring 里的例子。
卡住 8 分钟以上再按 ai-guide.md 的模板提问;拿到提示后自己重写。
预测练习(先在纸上写出答案,再到 REPL 验证):
>>> from typing import Literal, TypedDict, NotRequired
>>> class U(TypedDict):
... name: str
... email: NotRequired[str]
>>> print(U(name="a") == {"name": "a"}, isinstance({"name": "a"}, dict))
>>> Mode = Literal["r", "w"]
>>> print(Mode.__args__)
>>> def f[T](x: T) -> T: return x
>>> print(f.__type_params__)
>>> class Box[T]: ...
>>> print(Box[int], Box[int]().__class__ is Box)
"""
from collections.abc import Callable, Sequence
from typing import (
Final,
Literal,
NotRequired,
Protocol,
Self,
TypedDict,
TypeIs,
Unpack,
overload,
runtime_checkable,
)
# 3.14 的注解是延迟求值的:不需要 `from __future__ import annotations`,
# 前向引用(下面 Json 里的 "Json")直接写字符串或裸名字都可以。
MAX_STACK: Final = 100
"""Final:常量。类型检查器会拒绝对它重新赋值。"""
# ---- 任务 1:递归类型别名 + TypeIs ----
type Json = dict[str, "Json"] | list["Json"] | str | int | float | bool | None
# 记录每个被 @timed 装饰的函数最近一次耗时(秒)
LAST_DURATIONS: dict[str, float] = {}
def is_json(value: object) -> TypeIs[Json]:
"""判断 value 能不能直接被 json.dumps 序列化(只认 Json 别名里的那几种)。
规则:None/str/int/float/bool 是 True;list 要求每个元素都是 Json;
dict 要求**键是 str** 且值都是 Json;其他(set、对象、bytes……)是 False。
例:is_json({"a": [1, None]}) -> True;is_json({1: "a"}) -> False
例:is_json({"a", "b"}) -> False
"""
# TODO: 在这里实现(递归;用 isinstance 判断容器)
raise NotImplementedError
# ---- 任务 2:PEP 695 泛型类 + Self ----
class Stack[T]:
"""后进先出的栈。push 返回 self(Self 类型),所以可以链式调用。
例:Stack[int]().push(1).push(2).pop() -> 2
"""
def __init__(self) -> None:
"""创建一个空栈。"""
# TODO: 在这里实现(提示:self._items: list[T] = [])
raise NotImplementedError
def push(self, item: T) -> Self:
"""压入一个元素并返回自己,便于链式调用。"""
# TODO: 在这里实现
raise NotImplementedError
def pop(self) -> T:
"""弹出并返回栈顶;空栈时抛 IndexError("栈是空的")。"""
# TODO: 在这里实现
raise NotImplementedError
def peek(self) -> T:
"""看一眼栈顶但不弹出;空栈时抛 IndexError("栈是空的")。"""
# TODO: 在这里实现
raise NotImplementedError
def __len__(self) -> int:
"""栈里的元素个数(这样 len(stack) 才可用)。"""
# TODO: 在这里实现
raise NotImplementedError
def is_empty(self) -> bool:
"""是否为空栈。例:Stack[str]().is_empty() -> True"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 3:泛型函数 ----
def pairwise[T](xs: Sequence[T]) -> list[tuple[T, T]]:
"""相邻两两配对。参数用宽泛的 Sequence,返回用具体的 list。
例:pairwise([1, 2, 3]) -> [(1, 2), (2, 3)]
例:pairwise(["a"]) -> []
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 4:ParamSpec 装饰器 ----
def timed[**P, R](fn: Callable[P, R]) -> Callable[P, R]:
"""装饰器:把 fn 的耗时(秒)记到 LAST_DURATIONS[fn.__name__],返回值原样返回。
要求:用 functools.wraps 保留 __name__/__doc__(自己 import);
用 time.perf_counter() 计时;异常也要能穿透(不要吞掉)。
`**P`/`R` 的作用是保留原函数签名——在 VS Code 里把光标放到被装饰的函数上,
Pylance 显示的参数应该和原函数一致(相当于 reveal_type 的效果)。
例:@timed 装饰 add(a, b) 后,add(1, 2) -> 3,且 LAST_DURATIONS["add"] >= 0
"""
# TODO: 在这里实现(写一个内层 wrapper(*args: P.args, **kwargs: P.kwargs) -> R)
raise NotImplementedError
# ---- 任务 5:TypedDict + NotRequired + Unpack ----
class User(TypedDict):
"""一个"形状固定的字典":email 可有可无。"""
name: str
age: int
email: NotRequired[str]
def make_user(**kwargs: Unpack[User]) -> User:
"""构造 User:name 去掉首尾空白;age 为负时抛 ValueError("age 不能为负")。
email 没传就**不要**放进结果字典(这才是 NotRequired 的意思,
不是放一个 None 进去)。
例:make_user(name=" 小明 ", age=15) -> {"name": "小明", "age": 15}
例:make_user(name="a", age=1, email="a@b.com") 结果里含 email
"""
# TODO: 在这里实现
raise NotImplementedError
def describe_user(user: User) -> str:
"""一句话描述用户;有 email 就带上(用 .get 处理 NotRequired 的键)。
例:describe_user({"name": "小明", "age": 15}) -> "小明(15 岁)"
例:describe_user({"name": "a", "age": 1, "email": "x@y.z"}) -> "a(1 岁,x@y.z)"
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 6:Literal ----
type Mode = Literal["r", "w"]
def open_mode(mode: Mode) -> str:
"""把打开模式翻译成中文。Literal 让类型检查器在编译期就拦住 "rb"。
"r" -> "只读";"w" -> "只写";运行时收到别的值抛
ValueError(f"不支持的模式:{mode!r}")(类型检查器管不到运行时的脏数据)。
例:open_mode("r") -> "只读"
"""
# TODO: 在这里实现(推荐 match/case)
raise NotImplementedError
# ---- 任务 7:overload ----
@overload
def parse(x: str) -> int: ...
@overload
def parse(x: bytes) -> str: ...
def parse(x: str | bytes) -> int | str:
"""重载:给 str 返回 int(十进制解析),给 bytes 返回 str(utf-8 解码)。
其他类型抛 TypeError("parse 只接受 str 或 bytes")。
注意上面两个 @overload 只是给类型检查器看的声明,真正的实现只有这一个。
例:parse("42") -> 42;parse(b"hi") -> "hi"
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 8:泛型 Protocol ----
@runtime_checkable
class Repository[T](Protocol):
"""结构化子类型:谁有这三个方法,谁就算实现了 Repository[T]。
@runtime_checkable 只让 isinstance 检查"方法名在不在",不检查签名。
"""
def add(self, item: T) -> None:
"""加一个元素。"""
...
def get(self, index: int) -> T | None:
"""按下标取,越界返回 None。"""
...
def list_all(self) -> list[T]:
"""返回全部元素的副本。"""
...
class ListRepository[T]:
"""用列表实现 Repository[T](注意:不需要写 `(Repository[T])` 也算实现)。
例:r = ListRepository[int](); r.add(1); r.list_all() -> [1]
"""
def __init__(self) -> None:
"""创建空仓储。"""
# TODO: 在这里实现(提示:self._items: list[T] = [])
raise NotImplementedError
def add(self, item: T) -> None:
"""加一个元素。"""
# TODO: 在这里实现
raise NotImplementedError
def get(self, index: int) -> T | None:
"""按下标取,越界返回 None(不要抛 IndexError)。"""
# TODO: 在这里实现
raise NotImplementedError
def list_all(self) -> list[T]:
"""返回**副本**,调用方改它不影响仓储内部。"""
# TODO: 在这里实现
raise NotImplementedError
|
w2_04_pydantic_models.py
| exercises/week2/w2_04_pydantic_models.py |
|---|
| """Week 2 · M8a · Pydantic(一):模型、字段约束、校验器
学习目标:会用 BaseModel 做"外部输入的边界校验";会写 Field 约束、
field_validator/model_validator、ConfigDict(extra="forbid"/strict=True);
会读 ValidationError.errors() 的 loc/msg/type 并转成人能看懂的话。
用法:uv run pytest exercises/week2/test_w2_04_pydantic_models.py -q
做法:类与字段声明已经给好,你要做的是:
1) 把 TODO 注释里要求的 Field 约束补到字段上(或用 Annotated[T, Field(...)]);
2) 把 @field_validator / @model_validator 写出来;
3) 把函数体实现出来。不要改字段名字和类型。
卡住 8 分钟以上再按 ai-guide.md 的模板提问;拿到提示后自己重写。
预测练习(先在纸上写出输出,再到 REPL 验证):
>>> from pydantic import BaseModel, Field, ValidationError
>>> class Item(BaseModel):
... name: str = Field(min_length=1)
... qty: int = Field(gt=0)
... tags: list[str] = []
>>> print(Item(name="a", qty="3")) # qty 是 int 还是 str?
>>> try: Item(name="", qty=0)
... except ValidationError as e: print(len(e.errors()), [x["loc"] for x in e.errors()])
>>> a = Item(name="a", qty=1); b = Item(name="b", qty=1)
>>> a.tags.append("x"); print(b.tags) # 共享了吗?(Pydantic 会深拷贝默认值)
"""
from datetime import datetime
from decimal import Decimal
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, ValidationError
# 实现时你还需要自己 import:Annotated、Field、field_validator、model_validator、Self
# ---- 任务 1:Note ----
class Note(BaseModel):
"""一条笔记。
要求(用 Field 补到字段上):
- title:1–100 字,去掉首尾空白后仍不能为空
- tags:最多 5 个(Field(max_length=5))——注意这个长度检查发生在**去重之前**,
所以 ["a"] * 6 会报错,这是 Pydantic 的一个小陷阱
- created_at:默认"现在",且必须带时区(UTC)
还要写一个 @field_validator("tags") 把标签统一成小写、去掉空白、按出现顺序去重。
例:Note(title="标题", tags=["Python", "python", " AI "]).tags -> ["python", "ai"]
例:Note(title="t").created_at.tzinfo is not None -> True
"""
# extra="forbid":多写一个字段就报错,早发现拼错的键名
# str_strip_whitespace=True:所有 str 字段自动去首尾空白
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
# TODO: title 加 Field(min_length=1, max_length=100)
title: str
body: str = ""
# TODO: tags 加 Field(max_length=5);注意可变默认值 [] 在 Pydantic 里是安全的
tags: list[str] = []
# TODO: created_at 加 Field(default_factory=...),用 datetime.now(UTC)
created_at: datetime
priority: Literal["low", "normal", "high"] = "normal"
# TODO: 写 @field_validator("tags") 实现"小写 + 去空白 + 去空串 + 顺序去重"
# 签名参考:
# @field_validator("tags")
# @classmethod
# def _normalize_tags(cls, v: list[str]) -> list[str]: ...
# ---- 任务 2:嵌套模型 ----
class Address(BaseModel):
"""收件地址。
要求:city 至少 1 字;zipcode 必须是 6 位数字(Field(pattern=...),记得写 r"")。
例:Address(city="北京", street="中关村 1 号", zipcode="100080").zipcode -> "100080"
"""
model_config = ConfigDict(extra="forbid")
# TODO: city 加 Field(min_length=1)
city: str
street: str = ""
# TODO: zipcode 加 Field(pattern=r"^\d{6}$")
zipcode: str
class Profile(BaseModel):
"""用户资料:把 Address 嵌进来。
要求:age 用 Annotated[int, Field(ge=0, le=150)] 写法(自己 import Annotated 和
Field)。
注意 nickname: str | None = None 才是"可省略";只写 str | None 仍然是必填。
例:Profile(name="小明", age=15, address={"city": "北京", "zipcode": "100080"})
会自动把那个 dict 变成 Address 实例
"""
model_config = ConfigDict(extra="forbid")
name: str
# TODO: age 改成 Annotated[int, Field(ge=0, le=150)]
age: int
address: Address
nickname: str | None = None
# ---- 任务 3:跨字段校验 ----
class Meeting(BaseModel):
"""会议:结束时间必须晚于开始时间。
要求:写 @model_validator(mode="after"),不满足时 raise ValueError("start 必须早于
end")。
mode="after" 的校验器拿到的是**已经建好的模型实例**,返回 self。
例:Meeting(title="周会", start=..., end=...) start >= end 时抛 ValidationError
"""
title: str
start: datetime
end: datetime
# TODO: 写 @model_validator(mode="after") def _check_order(self) -> Self: ...
# (Self 来自 typing)
# ---- 任务 4:严格模式 + Decimal ----
class Money(BaseModel):
"""金额。strict=True 关掉"宽松解析":字符串 "1.5" 不再被悄悄转成 Decimal。
要求:amount 加 Field(gt=0);currency 加 Field(pattern=r"^[A-Z]{3}$")。
例:Money(amount=Decimal("1.5"), currency="CNY") 可以
例:Money(amount="1.5", currency="CNY") 抛 ValidationError(strict 模式)
"""
model_config = ConfigDict(strict=True)
# TODO: amount 加 Field(gt=0)
amount: Decimal
# TODO: currency 加 Field(pattern=r"^[A-Z]{3}$")
currency: str
# ---- 任务 5:收集错误 ----
def format_errors(exc: ValidationError, index: int) -> list[str]:
"""把一个 ValidationError 变成人能读的字符串列表。
每条格式:f"第 {index} 条 {loc}: {msg}",其中 loc 是把 err["loc"] 里的各段
用 "." 连起来(元素可能是 int,要先 str()),msg 直接用 err["msg"]。
例:一条 title 太短的错误 -> ["第 0 条 title: String should have at least 1
character"]
"""
# TODO: 在这里实现(遍历 exc.errors())
raise NotImplementedError
def parse_notes(raw: list[dict[str, Any]]) -> tuple[list[Note], list[str]]:
"""批量解析笔记:好的收进列表,坏的把错误记下来,**不要**因为一条坏数据就整批失败。
返回 (成功的 Note 列表, 错误描述列表);错误描述用 format_errors 生成。
例:parse_notes([{"title": "a"}, {"title": ""}]) -> (1 条 Note, 1 条错误)
"""
# TODO: 在这里实现(try/except ValidationError,逐条处理)
raise NotImplementedError
# ---- 任务 6:model_copy ----
def promote(note: Note) -> Note:
"""返回一个 priority 为 "high" 的副本,原对象保持不变。
要求用 note.model_copy(update={...})(注意:model_copy 不会重新校验)。
例:promote(Note(title="t")).priority -> "high"
"""
# TODO: 在这里实现
raise NotImplementedError
__all__ = [
"Address",
"Meeting",
"Money",
"Note",
"Profile",
"format_errors",
"parse_notes",
"promote",
]
|
w2_05_pydantic_schema_settings.py
| exercises/week2/w2_05_pydantic_schema_settings.py |
|---|
| """Week 2 · M8b · Pydantic(二):序列化、JSON Schema、Settings、TypeAdapter
学习目标:会用 model_dump(mode="json") 处理 datetime/Decimal;会读 model_json_schema()
(这就是"函数 → 结构化参数说明"的机制);会用判别联合与 TypeAdapter;
会用 pydantic-settings 从环境变量/.env 读配置。
用法:uv run pytest exercises/week2/test_w2_05_pydantic_schema_settings.py -q
做法:把每个 TODO 换成你的实现。模型(Click/KeyPress/ApiNote/Article/Doc)已经写全了,
读懂它们;你要写的是函数体 + Settings 的一个配置项。
卡住 8 分钟以上再按 ai-guide.md 的模板提问;拿到提示后自己重写。
预测练习(先在纸上写出输出,再到 REPL 验证):
>>> from datetime import UTC, datetime
>>> from pydantic import BaseModel
>>> class E(BaseModel):
... at: datetime
... note: str | None = None
>>> e = E(at=datetime(2026, 9, 14, tzinfo=UTC))
>>> print(e.model_dump()) # at 是什么类型?
>>> print(e.model_dump(mode="json")) # 现在呢?
>>> print(e.model_dump(mode="json", exclude_none=True))
>>> import json; print(json.dumps(e.model_dump())) # 这一行会怎样?
>>> print(E.model_json_schema()["required"])
"""
from datetime import datetime
from decimal import Decimal
from pathlib import Path
from typing import Annotated, Any, Literal
from pydantic import BaseModel, ConfigDict, Field, computed_field
from pydantic_settings import BaseSettings, SettingsConfigDict
# 实现时你还需要自己 import:TypeAdapter(以及计时用的 time)
# ---- 已给:判别联合的两个成员 ----
class Click(BaseModel):
"""鼠标点击事件。kind 是"判别字段"(discriminator)。"""
kind: Literal["click"] = "click"
x: int
y: int
class KeyPress(BaseModel):
"""键盘事件。"""
kind: Literal["key"] = "key"
key: str
# Annotated[... , Field(discriminator="kind")]:Pydantic 先看 kind,
# 再决定用哪个模型解析,
# 比"两个都试一遍"既快又能给出准确的错误信息。
type Event = Annotated[Click | KeyPress, Field(discriminator="kind")]
# ---- 任务 1:判别联合 + TypeAdapter ----
def parse_event(data: dict[str, Any]) -> Click | KeyPress:
"""按 kind 把一个 dict 解析成 Click 或 KeyPress。
要求:用 TypeAdapter(Event),并且把 adapter 建成**模块级常量**复用
(每次新建 TypeAdapter 都要重新编译校验器,很浪费):
EVENT_ADAPTER: TypeAdapter[Event] = TypeAdapter(Event)
kind 缺失或不认识时,让 Pydantic 自己抛 ValidationError(不要自己转成别的异常)。
例:parse_event({"kind": "click", "x": 1, "y": 2}) -> Click(x=1, y=2)
例:parse_event({"kind": "key", "key": "a"}) -> KeyPress(key="a")
"""
# TODO: 在这里实现
raise NotImplementedError
def parse_events(raw: list[dict[str, Any]]) -> list[Click | KeyPress]:
"""一次解析一批事件(用 TypeAdapter(list[Event]),同样做成模块级常量)。
例:parse_events([{"kind": "key", "key": "a"}]) -> [KeyPress(key="a")]
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 已给:一个"对外输出"的模型 ----
class ApiNote(BaseModel):
"""要发给前端的笔记。
- alias="createdAt":输入和输出都用驼峰(前端习惯),Python 侧仍是蛇形
- populate_by_name=True:也允许用 Python 字段名 created_at 传值
- @computed_field:算出来的字段,会出现在 model_dump 里
"""
model_config = ConfigDict(populate_by_name=True)
title: str
created_at: datetime = Field(alias="createdAt")
price: Decimal
note_id: int | None = None
@computed_field
@property
def summary(self) -> str:
"""一句话摘要(计算字段)。"""
return f"{self.title}({self.price} 元)"
def dump_for_api(model: BaseModel) -> dict[str, Any]:
"""把任意模型变成"可以直接 json.dumps 的 dict"。
要求一次给齐三个参数:mode="json"(datetime -> ISO 字符串、Decimal -> 字符串)、
exclude_none=True(None 的字段不发出去)、by_alias=True(用驼峰键名)。
例:dump_for_api(ApiNote(title="t", createdAt=..., price=Decimal("1.5")))
-> {"title": "t", "createdAt": "2026-09-14T00:00:00Z", "price": "1.5", ...}
且结果里没有 note_id
"""
# TODO: 在这里实现(一行)
raise NotImplementedError
# ---- 已给:一个用来看 JSON Schema 的模型 ----
class Article(BaseModel):
"""文章。注意每个字段的 description 都会进 JSON Schema。"""
title: str = Field(description="文章标题")
views: int = Field(default=0, description="阅读量")
author: str | None = Field(default=None, description="作者,可省略")
tags: list[str] = Field(default_factory=list, description="标签")
def schema_summary(model: type[BaseModel]) -> dict[str, Any]:
"""从 model_json_schema() 里提取一份"人和机器都能读"的摘要。
返回结构:
{
"required": [必填字段名, ...], # 直接取 schema.get("required", [])
"fields": {
字段名: {
"type": ..., # 有 "type" 就用它;否则把 "anyOf" 里各项的
# type 用 "|" 连起来(如 "string|null");
# 都没有就用 "unknown"
"description": ..., # 没有就空字符串
"default": ..., # 没有就 None
},
...
},
}
例:schema_summary(Article)["required"] -> ["title"]
例:schema_summary(Article)["fields"]["author"]["type"] -> "string|null"
"""
# TODO: 在这里实现(先 print 一下 model_json_schema() 看清结构再写)
raise NotImplementedError
# ---- 任务 3:Settings ----
class Settings(BaseSettings):
"""从环境变量 / .env 读配置。
要求:给 SettingsConfigDict 加上 env_prefix="KB_",这样 db_path 读的是
环境变量 KB_DB_PATH、log_level 读的是 KB_LOG_LEVEL(大小写不敏感)。
例:设了 KB_LOG_LEVEL=DEBUG 后 Settings().log_level -> "DEBUG"
例:什么都不设时 Settings().db_path -> Path("kb.db")
"""
# TODO: 加 env_prefix="KB_"
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
db_path: Path = Path("kb.db")
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
def describe_settings(settings: Settings) -> str:
"""一句话描述配置,用来在 CLI 启动时打日志。
格式:f"db={settings.db_path} level={settings.log_level}"
例:describe_settings(Settings()) -> "db=kb.db level=INFO"
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 4:TypeAdapter 校验一批数据 ----
class Doc(BaseModel):
"""一条文档记录(用来比较两种校验方式)。"""
id: int
title: str
created_at: datetime
def validate_docs(raw: list[dict[str, Any]]) -> list[Doc]:
"""用 TypeAdapter(list[Doc]) 一次校验整批(同样用模块级常量复用)。
例:validate_docs([{"id": 1, "title": "a", "created_at": "2026-09-14T00:00:00Z"}])
-> [Doc(id=1, ...)]
"""
# TODO: 在这里实现
raise NotImplementedError
def validate_docs_one_by_one(raw: list[dict[str, Any]]) -> list[Doc]:
"""逐条 Doc.model_validate(...),结果应该和 validate_docs 完全一样。
例:两个函数对同一批输入返回相等的列表
"""
# TODO: 在这里实现
raise NotImplementedError
def time_validation(raw: list[dict[str, Any]], repeat: int = 3) -> tuple[float, float]:
"""把两种方式各跑 repeat 遍,返回 (TypeAdapter 用的秒数, 逐条用的秒数)。
用 time.perf_counter() 计时。repeat <= 0 时抛 ValueError("repeat 必须为正")。
跑完自己看一眼:数据量小的时候差别不大,上千条时 TypeAdapter 明显更快。
例:time_validation(raw, repeat=1) -> (0.0012, 0.0031)(具体数字每次都不同)
"""
# TODO: 在这里实现
raise NotImplementedError
|
w2_06_logging_config.py
| exercises/week2/w2_06_logging_config.py |
|---|
| """Week 2 · M9 · logging 与配置(幂等的 setup_logging、JSON 日志、配置优先级)
学习目标:库代码只 getLogger、入口才配置 handler;会写 JSON Formatter;
知道 print 与 logging 的区别;会用 tomllib + Pydantic 读配置文件;
会按"命令行 > 环境变量 > .env > 默认值"的优先级解析配置。
用法:uv run pytest exercises/week2/test_w2_06_logging_config.py -q
做法:把每个 TODO 换成你的实现。注意 setup_logging 必须**幂等**:
重复调用不能让同一条日志打两遍——这是本周陷阱清单里的一条。
卡住 8 分钟以上再按 ai-guide.md 的模板提问;拿到提示后自己重写。
预测练习(先在纸上写出输出,再到 REPL 验证):
>>> import logging
>>> log = logging.getLogger("demo")
>>> print(log.level, log.getEffectiveLevel(), log.parent.name)
>>> logging.basicConfig(level=logging.INFO)
>>> log.info("hi") # 打出来了吗?格式是什么?
>>> logging.basicConfig(level=logging.DEBUG)
>>> log.debug("again") # basicConfig 第二次调用有效吗?
>>> root = logging.getLogger(); print(len(root.handlers))
>>> try: 1 / 0
... except ZeroDivisionError as e: log.error(e); log.exception(e) # 差别在哪?
"""
import logging
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
# 实现时你还需要自己 import:json、os、tomllib、
# logging.handlers.RotatingFileHandler、rich.logging.RichHandler
# 每个模块顶部一个 logger,名字用 __name__——这样日志里能看出是谁打的。
# 库代码**只做这一步**,绝不 basicConfig、绝不加 handler。
logger = logging.getLogger(__name__)
# 用来标记"这个 handler 是我加的",好让 setup_logging 幂等
MANAGED_FLAG = "_w2_managed"
# ---- 任务 1:JSON Formatter ----
class JsonFormatter(logging.Formatter):
"""把一条日志格式化成一行 JSON(机器可解析,方便以后喂给日志系统)。
输出的键固定为:
time —— 用 self.formatTime(record, "%Y-%m-%dT%H:%M:%S")
level —— record.levelname
logger —— record.name
message —— record.getMessage()(注意:不是 record.msg,要用 getMessage
才会把 logger.info("a=%s", 1) 里的参数套进去)
如果 record.exc_info 非空,再加一个键 exc_info,值是
self.formatException(record.exc_info)。
必须 ensure_ascii=False,否则中文会变成 \\uXXXX。
例:json.loads(JsonFormatter().format(record))["message"] -> "a=1"
"""
def format(self, record: logging.LogRecord) -> str:
"""把一条 LogRecord 变成一行 JSON 字符串。"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 2:幂等的 setup_logging ----
def setup_logging(
level: str = "INFO",
log_file: Path | None = None,
json: bool = False,
) -> logging.Logger:
"""配置**根** logger 并返回它。只应该被入口(CLI main)调用一次。
要求:
1. level 大小写不敏感("debug" 也行);不认识的级别抛
ValueError(f"未知日志级别:{level}")。可用 logging.getLevelNamesMapping()。
2. 控制台 handler 用 rich.logging.RichHandler(漂亮、带颜色)。
3. log_file 不为 None 时再加一个 RotatingFileHandler
(maxBytes=1_000_000, backupCount=3, encoding="utf-8");
父目录不存在要先 mkdir(parents=True, exist_ok=True)。
4. json=True 时两个 handler 都用 JsonFormatter,否则用普通 Formatter
(建议 "%(levelname)s %(name)s: %(message)s")。
5. **幂等**:每次调用先把上一次自己加的 handler 摘掉并 close()
(靠 getattr(h, MANAGED_FLAG, False) 认领;新加的 handler 上
setattr(h, MANAGED_FLAG, True))。
不要动别人加的 handler——pytest 的 caplog 就是往根 logger 加 handler 的,
全清掉的话测试就抓不到日志了。
例:setup_logging("DEBUG") 后 logging.getLogger().level -> 10
例:连续调用两次 setup_logging(log_file=p),自己加的 handler 始终是 2 个
"""
# TODO: 在这里实现
raise NotImplementedError
def managed_handlers() -> list[logging.Handler]:
"""返回根 logger 上"由 setup_logging 加的"那些 handler(测试和调试用)。
例:setup_logging() 后 len(managed_handlers()) -> 1
"""
# TODO: 在这里实现(用 getattr(h, MANAGED_FLAG, False) 过滤)
raise NotImplementedError
# ---- 任务 3:用 logger 而不是 print ----
def divide(a: float, b: float) -> float | None:
"""相除。成功时 logger.info,除零时 logger.exception 并返回 None。
要求:
- 成功:logger.info("divide %s / %s = %s", a, b, result) —— 用 %s 占位参数,
不要自己 f-string 拼(这样日志系统才能拿到结构化参数,也省掉不打印时的开销)
- 失败:except ZeroDivisionError: logger.exception("除以零:%s / %s", a, b)
然后 return None。注意是 exception 不是 error——error 会丢掉堆栈。
例:divide(6, 3) -> 2.0;divide(1, 0) -> None
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 4:配置文件 ----
class AppConfig(BaseModel):
"""应用配置(从 TOML 读出来后用它校验)。"""
model_config = ConfigDict(extra="forbid")
name: str = Field(min_length=1)
level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
max_items: int = Field(default=100, gt=0)
def load_config(path: Path) -> AppConfig:
"""用 tomllib 读 TOML 文件,再交给 AppConfig 校验。
要求:tomllib 只吃二进制流,所以用 path.open("rb");
文件不存在时让 FileNotFoundError 自然抛出(不要吞掉,也不要转成别的异常);
TOML 语法错让 tomllib.TOMLDecodeError 抛出;字段不合法让 ValidationError 抛出。
例:内容为 'name = "kb"' 的文件 -> AppConfig(name="kb", level="INFO", max_items=100)
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 5:配置来源优先级 ----
def resolve_setting(cli_value: str | None, env_name: str, default: str) -> str:
"""按"命令行参数 > 环境变量 > 默认值"的优先级取一个配置值。
要求:cli_value 不是 None 就用它(哪怕是空串,命令行显式给的就算数);
否则看 os.environ 里的 env_name,**空串视为没设**;再否则用 default。
例:resolve_setting("DEBUG", "KB_LOG_LEVEL", "INFO") -> "DEBUG"
例:KB_LOG_LEVEL=WARNING 时 resolve_setting(None, "KB_LOG_LEVEL", "INFO")
-> "WARNING"
"""
# TODO: 在这里实现
raise NotImplementedError
|
w2_07_regex.py
| exercises/week2/w2_07_regex.py |
|---|
| r"""Week 2 · M10 · 正则表达式(命名分组、VERBOSE、sub 传函数、中文)
学习目标:会写命名分组并用 groupdict() 取值;分得清 search/match/fullmatch/findall;
会用非贪婪 .*?、字符类、前后查找与 re.VERBOSE 写可读的长正则;
会用 re.sub 做脱敏;知道什么时候**不该**用正则。
用法:uv run pytest exercises/week2/test_w2_07_regex.py -q
做法:把每个 TODO 换成你的实现。所有正则字面量都要写成原始字符串 r""(本周陷阱之一)。
建议在 regex101.com(方言选 Python)上边调边写。
注意:凡是 docstring 里出现反斜杠,都要给它加 r 前缀写成原始字符串,
否则 \d 这样的"非法转义"在 3.14 会告警。
卡住 8 分钟以上再按 ai-guide.md 的模板提问;拿到提示后自己重写。
预测练习(先在纸上写出输出,再到 REPL 验证):
>>> import re
>>> print(re.match(r"\d+", "abc 123")) # match 从哪里开始找?
>>> print(re.search(r"\d+", "abc 123").group())
>>> print(re.fullmatch(r"\d+", "123abc"))
>>> print(re.findall(r"a.*b", "a1b a2b")) # 贪婪
>>> print(re.findall(r"a.*?b", "a1b a2b")) # 懒惰
>>> print(re.sub(r"\s+", " ", "a b\tc"))
>>> print(re.split(r"[,;]\s*", "a, b;c"))
>>> print(len("我爱Python"), re.findall(r"[\u4e00-\u9fff]", "我爱Python"))
"""
import re
# TODO: 把这个占位正则换成能解析
# "2026-09-05 12:00:01 [ERROR] module: message" 的正则。
# 要求用命名分组 (?P<name>...),一共 5 个组:
# date(YYYY-MM-DD)、time(HH:MM:SS)、level(DEBUG|INFO|WARNING|ERROR|CRITICAL)、
# module(字母数字下划线点,不含空格)、message(该行剩下的全部,可以含冒号)
LOG_LINE_RE = re.compile(r"TODO")
# TODO: 用 re.VERBOSE 写一个可读的 URL 正则(带注释、分行),命名分组 scheme/host/path。
# 提示:VERBOSE 模式下空白被忽略、# 后面是注释,所以要匹配真正的 # 得写反斜杠加 #;
# 字符类里的 # 和空格仍然有效。
# 结构:(?P<scheme>https?) :// (?P<host>...) 可选端口 (?P<path>...) 可选 query 可选片段
URL_RE = re.compile(r"TODO", re.VERBOSE)
# ---- 任务 1 ----
def parse_log_line(line: str) -> dict[str, str] | None:
"""解析一行日志,返回 5 个命名分组组成的字典;不匹配返回 None。
用 fullmatch(整行必须匹配)+ groupdict()。
例:parse_log_line("2026-09-05 12:00:01 [ERROR] kb.cli: 打不开数据库: 权限不足")
-> {"date": "2026-09-05", "time": "12:00:01", "level": "ERROR",
"module": "kb.cli", "message": "打不开数据库: 权限不足"}
例:parse_log_line("随便一句话") -> None
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 2 ----
def extract_emails(text: str) -> list[str]:
"""按出现顺序提取所有邮箱地址。
规则:用户名部分允许字母数字和 . _ + -;域名至少有一个点。
例:extract_emails("a@b.com 和 x.y+z@mail.co.jp") -> ["a@b.com", "x.y+z@mail.co.jp"]
例:extract_emails("没有邮箱 @ 这里") -> []
"""
# TODO: 在这里实现(re.findall;注意不要让分组把结果变成元组)
raise NotImplementedError
# ---- 任务 3 ----
def extract_cn_phones(text: str) -> list[str]:
r"""提取中国大陆手机号:1 开头,第二位 3–9,共 11 位数字。
要求:前后不能再挨着数字(用 (?<!\d) 和 (?!\d),不要用
\b——数字和数字之间没有词边界)。
例:extract_cn_phones("打 13812345678 或 15900001111")
-> ["13812345678", "15900001111"]
例:extract_cn_phones("012345678901234") -> []
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 4 ----
def extract_dates(text: str) -> list[str]:
"""提取 ISO 8601 日期(YYYY-MM-DD),月份 01–12、日 01–31。
例:extract_dates("从 2026-09-14 到 2026-09-20") -> ["2026-09-14", "2026-09-20"]
例:extract_dates("2026-13-01 不是日期") -> []
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 5 ----
def normalize_whitespace(text: str) -> str:
r"""把所有连续空白(空格/制表/换行)压成一个空格,并去掉首尾空白。
例:normalize_whitespace(" a\tb\n\nc ") -> "a b c"
"""
# TODO: 在这里实现(一行 re.sub 就够)
raise NotImplementedError
# ---- 任务 6 ----
def strip_markdown(text: str) -> str:
r"""把 Markdown 变成纯文本,按这个顺序处理:
1. 用三个反引号围起来的代码块整体删掉(含围栏行本身,用 re.DOTALL + 非贪婪)
2. 行首的 #、##、### 等标题标记连同后面的空格删掉(用 re.MULTILINE)
3. **加粗** 去掉星号,只留文字
4. [文字](链接) 只留"文字"
5. 连续 3 个以上换行压成 2 个
6. 整体 strip()
例:strip_markdown("# 标题\n\n**粗**和[链接](http://a.b)") -> "标题\n\n粗和链接"
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 7 ----
def tokenize_zh_en(text: str) -> list[str]:
"""中英文混排分词:中文按**字**切,英文按**词**切,数字连成一个(含小数)。
标点和空白都丢掉。
例:tokenize_zh_en("我爱 Python 3.14!") -> ["我", "爱", "Python", "3.14"]
例:tokenize_zh_en("kb 是知识库") -> ["kb", "是", "知", "识", "库"]
"""
# TODO: 在这里实现(一个正则 + findall;用 | 把三种情况并起来)
raise NotImplementedError
# ---- 任务 8 ----
def mask_secrets(text: str) -> str:
r"""脱敏:日志/截图里绝不能出现密钥和完整手机号。
规则:
- sk- 开头 + 至少 8 位字母数字的密钥 -> 整个换成 "sk-***"
- 11 位手机号 -> 中间 4 位换成 ****(如 13812345678 -> 138****5678),
用 re.sub 的替换串引用分组(r"\1****\2")
例:mask_secrets("key=sk-abcd1234efgh 手机 13812345678")
-> "key=sk-*** 手机 138****5678"
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 9 ----
def parse_url(url: str) -> dict[str, str] | None:
"""用上面的 URL_RE 解析 URL,返回 {"scheme", "host", "path"};不匹配返回 None。
要求:整串必须匹配(fullmatch);没有路径时 path 是空串(不是 None);
query 和片段不进结果。
例:parse_url("https://cnb.cool/me/repo?tab=1")
-> {"scheme": "https", "host": "cnb.cool", "path": "/me/repo"}
例:parse_url("http://localhost:8000")
-> {"scheme": "http", "host": "localhost", "path": ""}
例:parse_url("ftp://a.b/c") -> None
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 10 ----
def count_words(text: str) -> dict[str, int]:
"""统计英文单词出现次数(忽略大小写),按次数降序、次数相同按字母升序返回。
这一题的重点是**反面教材**:分词用正则,但排序、计数用 dict 和 sorted
——不要什么都想用正则解决。
例:count_words("a b A c b") -> {"a": 2, "b": 2, "c": 1}
"""
# TODO: 在这里实现(re.findall(r"[A-Za-z]+", ...) + Counter 或 dict)
raise NotImplementedError
|
w2_08_files_sqlite.py
| exercises/week2/w2_08_files_sqlite.py |
|---|
| """Week 2 · M11 · 文件与序列化进阶 + SQLite
学习目标:会用 pathlib 遍历目录、算文件哈希、原子写 JSON、处理 BOM;
存时间一律用带时区的 UTC ISO 字符串;会用 sqlite3 建表建索引、
参数化查询(防注入)、with conn 事务、row_factory。
用法:uv run pytest exercises/week2/test_w2_08_files_sqlite.py -q
做法:把每个 TODO 换成你的实现。数据库测试全部用 tmp_path 下的临时文件,
绝不要往仓库里写 .db。
卡住 8 分钟以上再按 ai-guide.md 的模板提问;拿到提示后自己重写。
预测练习(先在纸上写出输出,再到 REPL 验证):
>>> import json, sqlite3
>>> from datetime import UTC, datetime
>>> print(json.dumps({"n": "中文"})) # 中文变成什么了?
>>> print(json.dumps({"n": "中文"}, ensure_ascii=False))
>>> print(datetime.now().tzinfo, datetime.now(UTC).isoformat()[-6:])
>>> conn = sqlite3.connect(":memory:")
>>> conn.execute("CREATE TABLE t(x TEXT)")
>>> conn.execute("INSERT INTO t VALUES (?)", ("a",))
>>> print(conn.execute("SELECT * FROM t").fetchall()) # 没 commit 也能查到吗?
>>> conn.row_factory = sqlite3.Row
>>> row = conn.execute("SELECT * FROM t").fetchone()
>>> print(row["x"], dict(row))
"""
import sqlite3
from collections.abc import Iterator, Sequence
from pathlib import Path
from typing import Any
# 实现时你还需要自己 import:csv、hashlib、json、os、re、
# datetime(UTC、datetime)、decimal(Decimal)
# 建表与索引:写成 IF NOT EXISTS,所以 migrate 可以反复跑(幂等)
SCHEMA = """
CREATE TABLE IF NOT EXISTS notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
body TEXT NOT NULL DEFAULT '',
tags TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_notes_created_at ON notes (created_at);
"""
# ---- 任务 1 ----
def iter_markdown_files(root: Path) -> Iterator[Path]:
"""递归找出 root 下所有 .md 文件,按路径排序后逐个 yield。
要求:跳过任何路径里含"以点开头的目录"的文件(.git/、.venv/ 里的不要);
后缀比较不分大小写(.MD 也算);root 不存在时抛
FileNotFoundError(f"目录不存在:{root}")。
例:iter_markdown_files(Path("research")) -> 逐个 Path
"""
# TODO: 在这里实现(提示:root.rglob("*")、p.suffix.lower()、p.relative_to(root)
# .parts)
raise NotImplementedError
# ---- 任务 2 ----
def file_sha256(path: Path, chunk_size: int = 65536) -> str:
"""算文件的 SHA-256 十六进制摘要,**分块读**(大文件不能一次读进内存)。
例:内容为 b"hello" 的文件 ->
"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
"""
# TODO: 在这里实现(hashlib.sha256() + while 循环读 chunk_size 字节)
raise NotImplementedError
# ---- 任务 3 ----
def json_default(obj: Any) -> Any:
"""给 json.dump 的 default= 用:把它不认识的类型转成能序列化的东西。
规则:datetime -> isoformat() 字符串;Decimal -> str;set/frozenset -> 排序后的
list;
其他类型抛 TypeError(f"不支持的类型:{type(obj).__name__}")。
例:json_default(Decimal("1.5")) -> "1.5"
例:json_default({"b", "a"}) -> ["a", "b"]
"""
# TODO: 在这里实现
raise NotImplementedError
def export_json(path: Path, notes: list[dict[str, Any]]) -> None:
"""把 notes 原子地写成 JSON 文件。
要求:
- **原子写**:先写同目录下的临时文件(比如 path.with_suffix(".tmp")),
再用 os.replace(tmp, path) 一步换上去——中途断电也不会留下半个文件
- json.dump(..., ensure_ascii=False, indent=2, default=json_default)
- 编码 utf-8;父目录不存在要先建
- 出错时不要留下临时文件(try/except 里删掉它)
例:export_json(p, [{"title": "中文", "at": datetime.now(UTC)}]) 后
json.loads(p.read_text("utf-8"))[0]["title"] -> "中文"
"""
# TODO: 在这里实现
raise NotImplementedError
# ---- 任务 4 ----
def read_csv_with_bom(path: Path) -> list[dict[str, str]]:
r"""读 Excel 导出的 CSV(带 BOM),返回每行一个 dict。
要点:编码用 "utf-8-sig"(否则第一个表头会带上不可见的 \ufeff);
open 要写 newline=""(否则 Windows 上会多出空行)。
例:表头 name,score 的文件 -> [{"name": "小明", "score": "90"}]
"""
# TODO: 在这里实现(csv.DictReader)
raise NotImplementedError
# ---- 任务 5 ----
def utc_now_iso() -> str:
"""当前时间的 UTC ISO 8601 字符串(**带时区**,存库一律用这个)。
例:utc_now_iso() -> "2026-09-14T02:03:04.567890+00:00"
例:datetime.fromisoformat(utc_now_iso()).tzinfo is not None -> True
"""
# TODO: 在这里实现(datetime.now(UTC).isoformat())
raise NotImplementedError
# ---- 任务 6 ----
def migrate(conn: sqlite3.Connection) -> None:
"""建表建索引,幂等(跑几遍都一样)。
要求:用 with conn: 包住 executescript(SCHEMA)(这样才有事务、才会提交);
另外把 PRAGMA journal_mode=WAL 打开(并发读写更稳)。
例:migrate(conn); migrate(conn) 不报错
"""
# TODO: 在这里实现
raise NotImplementedError
class SqliteNoteRepository:
"""用 SQLite 存笔记,实现第 1 周那个 Repository 协议的 add/get/list/search/delete。
每行长这样:{"id": 1, "title": "t", "body": "", "tags": ["a"], "created_at": "..."}
tags 在库里存成逗号分隔的字符串,读出来还原成 list[str]。
例:repo = SqliteNoteRepository(tmp_path / "kb.db"); repo.add("标题") -> 1
"""
def __init__(self, db_path: Path) -> None:
"""连上数据库,设置 row_factory = sqlite3.Row,并跑一次 migrate。
要求:父目录不存在先建;把连接存成 self.conn。
"""
# TODO: 在这里实现
raise NotImplementedError
def add(self, title: str, body: str = "", tags: Sequence[str] = ()) -> int:
"""插入一条笔记,返回新的 id(cursor.lastrowid)。
要求:**必须**用 ? 参数化(绝不能用 f-string 拼 SQL);
created_at 用 utc_now_iso();tags 用 ",".join(tags) 存。
例:repo.add("标题", tags=["python"]) -> 1
"""
# TODO: 在这里实现(with self.conn: 才会提交事务)
raise NotImplementedError
def get(self, note_id: int) -> dict[str, Any] | None:
"""按 id 取一条,不存在返回 None。
例:repo.get(999) -> None
"""
# TODO: 在这里实现
raise NotImplementedError
def list(self) -> list[dict[str, Any]]:
"""按 id 升序返回全部笔记。
例:repo.list() -> [{"id": 1, ...}, {"id": 2, ...}]
"""
# TODO: 在这里实现
raise NotImplementedError
def search(self, pattern: str, *, regex: bool = False) -> list[dict[str, Any]]:
"""在 title 和 body 里搜,返回命中的笔记(都不区分大小写)。
regex=False:用 SQL 的 LIKE '%pattern%'(参数化传 f"%{pattern}%")。
regex=True:把候选行取到 Python 侧,用 re.search(pattern, title + body,
re.IGNORECASE) 过滤——SQLite 默认不带正则函数。
例:repo.search("python") -> 标题或正文里含 python 的那些
例:repo.search(r"正则.*表达式", regex=True) -> 用正则匹配
"""
# TODO: 在这里实现
raise NotImplementedError
def delete(self, note_id: int) -> bool:
"""删一条,返回是否真的删掉了(看 cursor.rowcount)。
例:repo.delete(1) -> True;再删一次 -> False
"""
# TODO: 在这里实现
raise NotImplementedError
def close(self) -> None:
"""关闭连接(Windows 上不关的话临时目录可能删不掉)。"""
# TODO: 在这里实现
raise NotImplementedError
def _row_to_dict(self, row: sqlite3.Row) -> dict[str, Any]:
"""把一行 sqlite3.Row 变成 dict,并把 tags 字符串还原成 list[str]。
注意 "".split(",") 是 [""],不是 []——空标签要特殊处理。
例:tags 字段是 "a,b" -> {"tags": ["a", "b"], ...}
"""
# TODO: 在这里实现
raise NotImplementedError
|