Loading...
Loading...
Compare original and translation side by side
undefinedundefinedundefinedundefinedruffundefinedruffundefined
Run with:
```bash
ruff check --fix . # Lint and auto-fix
ruff format . # Format code
运行命令:
```bash
ruff check --fix . # 代码检查并自动修复
ruff format . # 格式化代码undefinedundefined
Alternative: Use `pyright` for faster checking.
```toml
[tool.pyright]
pythonVersion = "3.12"
typeCheckingMode = "strict"
替代方案:使用`pyright`进行更快的检查。
```toml
[tool.pyright]
pythonVersion = "3.12"
typeCheckingMode = "strict"undefinedundefined
**Classes and Functions:**
```python
**类与函数:**
```python
**Constants:**
```python
**常量:**
```pythonundefinedundefinedundefinedundefined
Use absolute imports exclusively:
```python
仅使用绝对导入:
```pythonundefinedundefineddef get_user(user_id: str) -> User:
"""Retrieve a user by their unique identifier."""
...def process_batch(
items: list[Item],
max_workers: int = 4,
on_progress: Callable[[int, int], None] | None = None,
) -> BatchResult:
"""Process items concurrently using a worker pool.
Processes each item in the batch using the configured number of
workers. Progress can be monitored via the optional callback.
Args:
items: The items to process. Must not be empty.
max_workers: Maximum concurrent workers. Defaults to 4.
on_progress: Optional callback receiving (completed, total) counts.
Returns:
BatchResult containing succeeded items and any failures with
their associated exceptions.
Raises:
ValueError: If items is empty.
ProcessingError: If the batch cannot be processed.
Example:
>>> result = process_batch(items, max_workers=8)
>>> print(f"Processed {len(result.succeeded)} items")
"""
...class UserService:
"""Service for managing user operations.
Provides methods for creating, retrieving, updating, and
deleting users with proper validation and error handling.
Attributes:
repository: The data access layer for user persistence.
logger: Logger instance for operation tracking.
Example:
>>> service = UserService(repository, logger)
>>> user = service.create_user(CreateUserInput(...))
"""
def __init__(self, repository: UserRepository, logger: Logger) -> None:
"""Initialize the user service.
Args:
repository: Data access layer for users.
logger: Logger for tracking operations.
"""
self.repository = repository
self.logger = loggerdef get_user(user_id: str) -> User:
"""通过唯一标识符获取用户。"""
...def process_batch(
items: list[Item],
max_workers: int = 4,
on_progress: Callable[[int, int], None] | None = None,
) -> BatchResult:
"""使用工作池并发处理项目。
配置指定数量的工作线程并发处理批量项目。可通过可选回调函数监控进度。
参数:
items: 待处理的项目列表,不能为空。
max_workers: 最大并发工作线程数,默认值为4。
on_progress: 可选回调函数,接收(已完成数, 总数)作为参数。
返回:
包含成功处理项目和失败项目(含关联异常)的BatchResult对象。
异常:
ValueError: 当items为空时抛出。
ProcessingError: 当批量处理无法执行时抛出。
示例:
>>> result = process_batch(items, max_workers=8)
>>> print(f"已处理 {len(result.succeeded)} 个项目")
"""
...class UserService:
"""用户操作管理服务。
提供用户创建、查询、更新和删除的方法,并包含完善的验证和错误处理机制。
属性:
repository: 用户持久化的数据访问层。
logger: 用于操作追踪的日志实例。
示例:
>>> service = UserService(repository, logger)
>>> user = service.create_user(CreateUserInput(...))
"""
def __init__(self, repository: UserRepository, logger: Logger) -> None:
"""初始化用户服务。
参数:
repository: 用户数据访问层。
logger: 用于操作追踪的日志实例。
"""
self.repository = repository
self.logger = loggerundefinedundefinedundefinedundefinedundefinedundefined
**CHANGELOG Format (Keep a Changelog):**
```markdown
**CHANGELOG格式(遵循Keep a Changelog规范):**
```markdownundefinedundefined