我正在对某些功能进行测试。我有一个使用数据库查询的函数。所以,我浏览了一些博客和文档,这些博客和文档说我们必须在内存或测试数据库中创建一个才能使用这些功能。下面是我的功能,
def already_exists(story_data,c):
# TODO(salmanhaseeb): Implement de-dupe functionality by checking if it already
# exists in the DB.
c.execute("""SELECT COUNT(*) from posts where post_id = ?""", (story_data.post_id,))
(number_of_rows,)=c.fetchone()
if number_of_rows > 0:
return True
return False
此函数命中生产数据库。我的问题是,在测试时,我创建了一个内存数据库并在那里填充我的值,我将查询该数据库(测试数据库)。但是我想测试我的already_exists()
函数,already_exists
从测试调用我的函数后,我的生产数据库将被命中。在测试此功能时如何使我的测试数据库命中?
问题是确保您的代码始终使用相同的数据库连接。然后您可以将其设置为适合当前环境的任何值。
与其将数据库连接从一个方法传递到另一个方法,不如使其成为单例更有意义。
def already_exists(story_data):
# Here `connection` is a singleton which returns the database connection.
connection.execute("""SELECT COUNT(*) from posts where post_id = ?""", (story_data.post_id,))
(number_of_rows,) = connection.fetchone()
if number_of_rows > 0:
return True
return False
或者connection
在每个类上做一个方法,already_exists
变成一个方法。无论如何,它应该是一种方法。
def already_exists(self):
# Here the connection is associated with the object.
self.connection.execute("""SELECT COUNT(*) from posts where post_id = ?""", (self.post_id,))
(number_of_rows,) = self.connection.fetchone()
if number_of_rows > 0:
return True
return False
但实际上你不应该自己滚动这段代码。相反,您应该使用ORM,例如SQLAlchemy,它会为您处理基本查询和连接管理。它只有一个连接,即“会话”。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy_declarative import Address, Base, Person
engine = create_engine('sqlite:///sqlalchemy_example.db')
Base.metadata.bind = engine
DBSession = sessionmaker(bind=engine)
session = DBSession()
然后你用它来进行查询。例如,它有一个exists
方法。
session.query(Post.id).filter(q.exists()).scalar()
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句