python - 未填充外键的表
问题描述
这个周末我一直在学习用于 python 的sqlite3 stdlib 模块。我对 SQL 完全陌生。
我试图编写一个测试外键的测试,但不幸的是,该segments
表似乎没有被填充,即fetchall()
没有返回任何结果。我查看了 db 文件,每个表的声明都存在,但points
似乎只填充了表。谁能提供一些见解?
import json
import attr
import random
import pytest
import datetime as dt
import typing as typ
import sqlite3 as sql
@attr.s
class Point:
x: float = attr.ib(default=random.random())
y: float = attr.ib(default=random.random())
def adapt_point(point: Point) -> str:
data = attr.asdict(point)
text = json.dumps(data)
return text
def convert_point(value: bytes) -> Point:
data = json.loads(value)
value = Point(**data)
return value
sql.register_adapter(Point, adapt_point)
sql.register_converter('point', convert_point)
@pytest.fixture
def connection():
con = sql.connect('tmpdb.sqlite', detect_types=sql.PARSE_DECLTYPES)
cur = con.cursor()
cur.execute(
"""
create table points(
id integer primary key,
pt point,
at timestamp
);
"""
)
cur.execute(
"""
create table segments(
id integer primary key,
start integer,
end integer,
at timestamp,
foreign key(start) references points(id),
foreign key(end) references points(id)
);
"""
)
con.commit()
yield con
con.close()
def test_insert_points(connection: sql.Connection):
"""
Test adding:
* datetime object using builtin 'timestamp' adapter and converter
* dataclass object using custom adapter and converter using :py:func:`attr.asdict`
:param connection: sqlite fixture
"""
cur = connection.cursor()
for i in range(5):
cur.execute(
"insert into points(id, pt, at) values (?, ?, ?)",
(i, Point(), dt.datetime.now())
)
connection.commit()
def test_insert_segments(connection: sql.Connection):
"""
Test adding:
* datetime object using builtin 'timestamp' adapter and converter
* foreign keys to other table
:param connection: sqlite fixture
"""
cur = connection.cursor()
for i in range(3):
cur.execute(
"insert into segments(id, start, end, at) values (?, ?, ?, ?)",
(i, random.randrange(5), random.randrange(5), dt.datetime.now())
)
connection.commit()
def test_retrieve_points_from_segment(connection: sql.Connection):
cur = connection.cursor()
cur.row_factory = sql.Row
cur.execute("select * from segments")
segs = cur.fetchall()
assert segs is not None, "THIS FAILS"
解决方案
推荐阅读
- python-3.x - 动态创建 GPIO PIN 变量
- reactjs - 使用 React JS 删除超过 2 天的 Firestore 文档
- css - Bootstrap 中 flex-fill 和 flex-grow-1 的区别
- jenkins - 通过管道(Jenkins)恢复vsphere的快照
- c - 在 C 中找到下一个素数?
- oauth-2.0 - Google API google-auth 身份验证永远等待,因为重定向到本地主机
- java - addAll 未解决的参考。由于接收器类型不匹配,以下候选均不适用
- c - 如何修复通过宏生成的代码的“标量初始化器周围的大括号”警告?
- state - Visualizer.pl 的问题
- asp.net-core - Identity Server 4 Discovery Endpoint 显示 Http 链接而不是 Https,即使从 Https 访问