Python 提供了处理数据、检查类型和封装常用功能的helper工具。在本节中，我们将介绍几个重要的工具，它们可以极大地改进您的代码。
1.类型Typing
虽然 Python 是一种动态语言，但拥有类型提示和类型检查器（如 mypy）可以在运行时避免多种类型不兼容问题。
例如，下面是 RedditETL 类中的 extract 方法，它显示了预期的输入和输出类型。
from typing import List
class RedditETL(SocialETL):
def extract(
self,
id: str,
num_records: int,
client: praw.Reddit,
) -> List[SocialMediaData]:
# Code
|
我们可以看到，提取方法需要三个输入：id（字符串类型）、num_records（整数类型）和客户端（piw.Reddit 类型），并返回一个 SocialMediaData 列表（我们将在数据类部分看到该类）。
Python 函数可以接受输入并产生输出，而输出本身就是函数（又称高阶函数）。我们还可以定义函数类型，如下所示。
from typing import Callable, List
# code
def transformation_factory(value: str) -> Callable[[List[SocialMediaData]], List[SocialMediaData]]:
factory = {
'sd': standard_deviation_outlier_filter,
'no_tx': no_transformation,
'rand': random_choice_filter,
}
return factory[value]
|
Callable[ [List[SocialMediaData]], List[SocialMediaData]] 表示函数。
- 第一个参数：定义函数的输入类型（[List[SocialMediaData]]）。Callable 的第一部分是一个列表，以考虑多种可能的输入。
- 第二个参数：定义函数的输出类型（List[SocialMediaData]）。
Mypy 是 Python 的静态类型检查器。我们可以运行 mypy来验证我们的代码是否遵守了定义的类型。例如，如果我们调用的函数类型与它期望的不同，mypy 就会抛出一个错误。2.数据类
数据类旨在将数据存储为数据类的对象。
优点
- 设计用于在 Python 中表示数据。
- 具有类型提示，并能通过集成开发环境完成代码。
- 可设置默认值。
- 可在创建对象时添加自定义数据处理。参见 post-init。
- 可使用冻结数据类模拟不变性。
- 可从其他数据类继承。
缺点
- 数据类是常规类，在创建过程中会产生一些开销。
- 对于简单的基于字典的数据来说，这是一种矫枉过正的做法。例如 [{'name': 'Tanya'}, {'name': 'Sophia'}] （{'name': 'Tanya'}, {'name': 'Sophia'}
在我们的代码中，我们使用 Dataclass 来表示从 Reddit 和 Twitter 获取的数据。
from dataclasses import asdict, dataclass
@dataclass
class RedditPostData:
""用于保存 reddit 帖子数据的数据类。
参数：
title （str）：reddit 帖子的标题。
score (int)：reddit 帖子的得分。
url (str)：reddit 帖子的 URL。
comms_num (int)：reddit 帖子的评论数。
created (str)：日期时间（字符串重写），表示 reddit
帖子的创建时间。
"""
title: str
score: int
url: str
comms_num: int
created: str
text: str
@dataclass
class TwitterTweetData:
""用于保存 twitter 发布数据的数据类。
参数：
text (str)：twitter 发布的文本。
"""
text: str
@dataclass
class SocialMediaData:
""用于保存社交媒体数据的数据类。
参数：
id （str）：社交媒体帖子的 ID。
text （str）：社交媒体文章的文本。
"""
id: str
source: str
social_data: RedditPostData | TwitterTweetData # social_data can be one of the Reddit or Twitter data types
|
3.上下文管理器
在建立与外部系统（如数据库、文件系统等）的连接时，我们应在连接完成后（或出现错误时）清理连接，以防止内存泄漏并释放系统资源。
在我们的示例中，我们需要在加载完成后（或出错时）关闭数据库连接。通常情况下，我们会使用 try...except...finally 代码块来完成这项工作，但无论我们在哪里连接数据库，都会产生重复的代码。
相反，我们可以创建上下文管理器，使用 with 块来调用。我们将为数据库连接创建一个上下文管理器，它会在成功或失败时自动关闭。我们可以使用 contextmanager 装饰器定义上下文管理器，如下所示。
from contextlib import contextmanager
class DatabaseConnection:
def __init__(
self, db_type: str = 'sqlite3', db_file: str = 'data/socialetl.db'
) -> None:
"用于连接数据库的类。
参数：
db_type （str，可选）：数据库类型。
默认为 "sqlite3"。
db_file （str，可选项）： 数据库文件：数据库文件。
默认为 "data/socialetl.db"。
"""
self._db_type = db_type
self._db_file = db_file
@contextmanager
def managed_cursor(self) -> Iterator[sqlite3.Cursor]:
"""创建托管数据库游标的函数。
产生：
sqlite3.Cursor：一个 sqlite3 游标。
"""
if self._db_type == 'sqlite3':
_conn = sqlite3.connect(self._db_file)
cur = _conn.cursor()
try:
yield cur
finally:
_conn.commit()
cur.close()
_conn.close()
def __str__(self) -> str:
return f'{self._db_type}://{self._db_file}'
db = DatabaseConnection()
with db.managed_cursor() as cur: # cursor and connection are open
cur.execute("YOUR SQL QUERY")
# cursor and connection are close
|
当我们使用 with db.managed_cursor() 作为 cur 子句时，代码如下
- 运行 managed_cursor 方法中的代码，直到 yield cur 一行。
- 将游标作为 cur 放在 with 代码块中。
- 在 with 代码块之后，代码控制返回到 managed_cursor 方法的 finally 部分，提交更改并关闭连接。
4.装饰器
装饰器为其他函数添加功能。例如
def log_metadata(func):
def log_wrapper():
print(f'Adding functionality to {func.__name__}')
# do some other thing
return func()
return log_wrapper
@log_metadata
def some_func():
print('do some thing')
|
在上面的示例中，我们用 log_metadata 函数装饰了 some_func 函数，为其添加了功能。log_metadata 函数可用作多个函数的装饰器。
在我们的项目中，我们使用 log_metadata 来记录函数名称、输入参数等信息，并将其转储到 log_metadata 表中，以跟踪数据的流向。inspect 模块可以帮助我们识别对象及其输入参数。
存储每个函数的输入、运行时间和函数名称可以帮助我们调试管道（又称数据脉络）中的问题。