Python 快速入门

Timescale 代码快速入门指南旨在帮助您将 Timescale 集成到您自己的程序中。它们使用您喜欢的编程语言来解释如何连接到 Timescale 数据库，创建和管理超表，以及摄取和查询数据。

本快速入门指南将引导您完成以下步骤

先决条件

在开始之前，请确保您已拥有

已安装 TimescaleDB。有关更多信息，请参阅安装文档。
已安装 psycopg2 库。有关更多信息，请参阅 psycopg2 文档。
可选Python 虚拟环境。

连接到 TimescaleDB

在本节中，您将使用 psycopg2 库创建与 TimescaleDB 的连接。此库是 Python 最流行的 PostgreSQL 库之一。它允许您高效且安全地执行原始 SQL 查询，并防止 SQL 注入等常见攻击。

导入 psycogpg2 库
```
import psycopg2
```
找到您的 TimescaleDB 凭据，并使用它们来编写 psycopg2 的连接字符串。
您将需要
- 密码
- 用户名
- 主机 URL
- 端口
- 数据库名称
将您的连接字符串变量编写为 libpq 连接字符串，使用此格式
```
CONNECTION = "postgres://username:password@host:port/dbname"
```
如果您使用的是 TimescaleDB 的托管版本，或者通常需要 SSL 连接，请改用此版本
```
CONNECTION = "postgres://username:password@host:port/dbname?sslmode=require"
```
或者，您可以按如下方式在连接字符串中指定每个参数
```
CONNECTION = "dbname=tsdb user=tsdbadmin password=secret host=host.com port=5432 sslmode=require"
```
警告
这种编写连接字符串的方法仅用于测试或开发目的。对于生产环境，请对密码、主机名和端口号等敏感详细信息使用环境变量。

使用 psycopg2 connect 函数创建新的数据库会话，并创建新的 cursor 对象以与数据库交互。

在您的 main 函数中，添加以下行


CONNECTION = "postgres://username:password@host:port/dbname"
with psycopg2.connect(CONNECTION) as conn:
    cursor = conn.cursor()
    # use the cursor to interact with your database
    # cursor.execute("SELECT * FROM table")

或者，您可以创建一个连接对象并在需要时传递该对象，例如打开游标以执行数据库操作


CONNECTION = "postgres://username:password@host:port/dbname"
conn = psycopg2.connect(CONNECTION)
cursor = conn.cursor()
# use the cursor to interact with your database
cursor.execute("SELECT 'hello world'")
print(cursor.fetchone())

创建关系表

在本节中，您将创建一个名为 sensors 的表，其中包含虚构传感器的 ID、类型和位置。此外，您还将创建一个名为 sensor_data 的超表，其中包含这些传感器的测量值。测量值包含时间、sensor_id、温度读数和传感器的 CPU 百分比。

编写一个字符串，其中包含用于创建关系表的 SQL 语句。此示例创建一个名为 sensors 的表，其中包含列 id、type 和 location


query_create_sensors_table = """CREATE TABLE sensors (
                                    id SERIAL PRIMARY KEY,
                                    type VARCHAR(50),
                                    location VARCHAR(50)
                                );
                                """

打开游标，执行您在上一步中创建的查询，并提交查询以使更改持久化。之后，关闭游标以进行清理
```
cursor = conn.cursor()
# see definition in Step 1
cursor.execute(query_create_sensors_table)
conn.commit()
cursor.close()
```

创建超表

创建关系表后，您可以创建超表。创建表和索引、更改表、插入数据、选择数据以及大多数其他任务都在超表上执行。

创建一个字符串变量，其中包含超表的 CREATE TABLE SQL 语句。请注意超表如何具有强制的时间列


# create sensor data hypertable
query_create_sensordata_table = """CREATE TABLE sensor_data (
                                        time TIMESTAMPTZ NOT NULL,
                                        sensor_id INTEGER,
                                        temperature DOUBLE PRECISION,
                                        cpu DOUBLE PRECISION,
                                        FOREIGN KEY (sensor_id) REFERENCES sensors (id)
                                    );
                                    """

制定一个 SELECT 语句，将 sensor_data 表转换为超表。您必须指定要转换为超表的表名，以及时间列的名称作为两个参数。有关更多信息，请参阅 create_hypertable 文档
```
query_create_sensordata_hypertable = "SELECT create_hypertable('sensor_data', by_range('time'));"
```
注意
by_range 维度构建器是 TimescaleDB 2.13 的新增功能。

使用连接打开游标，执行前几步的语句，提交您的更改，并关闭游标


cursor = conn.cursor()
cursor.execute(query_create_sensordata_table)
cursor.execute(query_create_sensordata_hypertable)
# commit changes to the database to make changes persistent
conn.commit()
cursor.close()

插入数据行

您可以通过几种不同的方式将数据插入到超表中。在本节中，您可以使用带有预处理语句的 psycopg2，也可以使用 pgcopy 以获得更快的插入速度。

此示例将元组列表或关系数据（称为 sensors）插入到名为 sensors 的关系表中。使用与数据库的连接打开游标，使用预处理语句来编写 INSERT SQL 语句，然后执行该语句


sensors = [('a', 'floor'), ('a', 'ceiling'), ('b', 'floor'), ('b', 'ceiling')]
cursor = conn.cursor()
for sensor in sensors:
  try:
    cursor.execute("INSERT INTO sensors (type, location) VALUES (%s, %s);",
                (sensor[0], sensor[1]))
  except (Exception, psycopg2.Error) as error:
    print(error.pgerror)
conn.commit()

可选或者，您可以将变量传递给 cursor.execute 函数，并将 SQL 语句 SQL 的编写与随其传递到预处理语句 data 中的数据分开


SQL = "INSERT INTO sensors (type, location) VALUES (%s, %s);"
sensors = [('a', 'floor'), ('a', 'ceiling'), ('b', 'floor'), ('b', 'ceiling')]
cursor = conn.cursor()
for sensor in sensors:
  try:
    data = (sensor[0], sensor[1])
    cursor.execute(SQL, data)
  except (Exception, psycopg2.Error) as error:
    print(error.pgerror)
conn.commit()

如果您选择使用 pgcopy，请使用 pip 安装 pgcopy 包，然后将此行添加到您的 import 语句列表中


from pgcopy import CopyManager

使用 PostgreSQL 提供的 generate_series 函数生成一些随机传感器数据。此示例共插入 480 行数据（4 个读数，每 5 分钟一次，持续 24 小时）。在您的应用程序中，这将是将您的时序数据保存到超表中的查询


# for sensors with ids 1-4
for id in range(1, 4, 1):
    data = (id,)
    # create random data
    simulate_query = """SELECT generate_series(now() - interval '24 hour', now(), interval '5 minute') AS time,
                            %s as sensor_id,
                            random()*100 AS temperature,
                            random() AS cpu;
                            """
    cursor.execute(simulate_query, data)
    values = cursor.fetchall()

定义您要将数据插入的表的列名。此示例使用先前创建的 sensor_data 超表。此超表由名为 time、sensor_id、temperature 和 cpu 的列组成。列名在名为 cols 的字符串列表中定义
```
cols = ['time', 'sensor_id', 'temperature', 'cpu']
```
创建 pgcopy CopyManager 的实例 mgr，并传递连接变量、超表名称和列名列表。然后使用 CopyManager 的 copy 函数，使用 pgcopy 快速将数据插入数据库。
```
mgr = CopyManager(conn, 'sensor_data', cols)
mgr.copy(values)
```
提交以持久化更改
```
conn.commit()
```

可选使用 pgcopy 将数据插入 TimescaleDB 的完整示例代码，使用来自四个传感器的传感器数据示例


# insert using pgcopy
def fast_insert(conn):
    cursor = conn.cursor()

    # for sensors with ids 1-4
    for id in range(1, 4, 1):
        data = (id,)
        # create random data
        simulate_query = """SELECT generate_series(now() - interval '24 hour', now(), interval '5 minute') AS time,
                                %s as sensor_id,
                                random()*100 AS temperature,
                                random() AS cpu;
                                """
        cursor.execute(simulate_query, data)
        values = cursor.fetchall()

        # column names of the table you're inserting into
        cols = ['time', 'sensor_id', 'temperature', 'cpu']

        # create copy manager with the target table and insert
        mgr = CopyManager(conn, 'sensor_data', cols)
        mgr.copy(values)

    # commit after all sensor data is inserted
    # could also commit after each sensor insert is done
    conn.commit()

可选您还可以检查插入是否成功


cursor.execute("SELECT * FROM sensor_data LIMIT 5;")
print(cursor.fetchall())

执行查询

本节介绍如何针对您的数据库执行查询。

第一个过程显示一个简单的 SELECT * 查询。对于更复杂的查询，您可以使用预处理语句来确保针对数据库安全地执行查询。

有关在 psycopg2 中正确使用占位符的更多信息，请参阅基本模块使用文档。有关如何在 psycopg2 中执行更复杂查询的更多信息，请参阅 psycopg2 文档。

定义您想在数据库上运行的 SQL 查询。此示例是一个简单的 SELECT 语句，查询先前创建的 sensor_data 表中的每一行。
```
query = "SELECT * FROM sensor_data;"
```

从现有数据库连接 conn 打开游标，然后执行您定义的查询


cursor = conn.cursor()
query = "SELECT * FROM sensor_data;"
cursor.execute(query)

要访问查询返回的所有结果行，请使用 pyscopg2 的结果检索方法之一，例如 fetchall() 或 fetchmany()。此示例逐行打印查询结果。请注意，fetchall() 的结果是元组列表，因此您可以相应地处理它们
```
cursor = conn.cursor()
query = "SELECT * FROM sensor_data;"
cursor.execute(query)
for row in cursor.fetchall():
    print(row)
cursor.close()
```
可选如果您想要字典列表，可以使用 DictCursor 定义游标
```
cursor = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
```
使用此游标，cursor.fetchall() 返回字典状对象的列表。

对于更复杂的查询，您可以使用预处理语句来确保针对数据库安全地执行查询。

使用预处理语句编写查询


# query with placeholders
cursor = conn.cursor()
query = """
           SELECT time_bucket('5 minutes', time) AS five_min, avg(cpu)
           FROM sensor_data
           JOIN sensors ON sensors.id = sensor_data.sensor_id
           WHERE sensors.location = %s AND sensors.type = %s
           GROUP BY five_min
           ORDER BY five_min DESC;
           """
location = "floor"
sensor_type = "a"
data = (location, sensor_type)
cursor.execute(query, data)
results = cursor.fetchall()

Python 快速入门

先决条件

连接到 TimescaleDB

连接到 TimescaleDB

警告

创建关系表

创建关系表

创建超表

创建超表

注意

插入数据行

使用 psycopg2 将行插入 TimescaleDB

使用 pgcopy 将行插入 TimescaleDB

执行查询

执行简单查询

使用预处理语句执行查询

Python 快速入门

先决条件

连接到 TimescaleDB

连接到 TimescaleDB

警告

创建关系表

创建关系表

创建超表

创建超表

注意

插入数据行

使用 psycopg2 将行插入 TimescaleDB

使用 pgcopy 将行插入 TimescaleDB

执行查询

执行简单查询

使用预处理语句执行查询

相关内容