本文整理汇总了Python中weedb.connect函数的典型用法代码示例。如果您正苦于以下问题:Python connect函数的具体用法?Python connect怎么用?Python connect使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了connect函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_transaction
def test_transaction(self):
# Create the database and schema
weedb.create(self.db_dict)
with weedb.connect(self.db_dict) as _connect:
# With sqlite, a rollback can roll back a table creation. With MySQL, it does not. So,
# create the table outside of the transaction. We're not as concerned about a transaction failing
# when creating a table, because it only happens the first time weewx starts up.
_connect.execute("""CREATE TABLE test1 ( dateTime INTEGER NOT NULL UNIQUE PRIMARY KEY, x REAL );""")
# We're going to trigger the rollback by raising a bogus exception. Be prepared to catch it.
try:
with weedb.Transaction(_connect) as _cursor:
for i in range(10):
_cursor.execute("""INSERT INTO test1 (dateTime, x) VALUES (?, ?)""", (i, i+1))
# Raise an exception:
raise Exception("Bogus exception")
except Exception:
pass
# Now make sure nothing is in the database
with weedb.connect(self.db_dict) as _connect:
with _connect.cursor() as _cursor:
_cursor.execute("SELECT dateTime, x from test1")
_row = _cursor.fetchone()
self.assertEqual(_row, None)
开发者ID:MLAB-project,项目名称:weewx,代码行数:26,代码来源:test_weedb.py
示例2: _create_table
def _create_table(archive_db_dict, archiveSchema, table):
"""Create a SQL table using a given archive schema.
archive_db_dict: A database dictionary holding the information
necessary to open the database.
archiveSchema: The schema to be used
table: The name of the table to be used within the database.
Returns:
A connection"""
# First try to create the database. If it already exists, an exception
# will be thrown.
try:
weedb.create(archive_db_dict)
except weedb.DatabaseExists:
pass
# List comprehension of the types, joined together with commas. Put
# the SQL type in backquotes, because at least one of them ('interval')
# is a MySQL reserved word
_sqltypestr = ', '.join(["`%s` %s" % _type for _type in archiveSchema])
_connect = weedb.connect(archive_db_dict)
try:
with weedb.Transaction(_connect) as _cursor:
_cursor.execute("CREATE TABLE %s (%s);" % (table, _sqltypestr))
except Exception, e:
_connect.close()
syslog.syslog(syslog.LOG_ERR, "archive: Unable to create table '%s' in database '%s': %s" %
(table, os.path.basename(archive_db_dict['database']), e))
raise
开发者ID:crmorse,项目名称:weewx-waterflow,代码行数:34,代码来源:archive.py
示例3: test_select
def test_select(self):
self.populate_db()
_connect = weedb.connect(self.db_dict)
_cursor = _connect.cursor()
_cursor.execute("SELECT dateTime, min FROM test1")
for i, _row in enumerate(_cursor):
self.assertEqual(_row[0], i)
# SELECT with wild card, using a result set
_result = _cursor.execute("SELECT * from test1")
for i, _row in enumerate(_result):
self.assertEqual(_row[0], i)
# Find a matching result set
_cursor.execute("SELECT dateTime, min FROM test1 WHERE dateTime = 5")
_row = _cursor.fetchone()
self.assertEqual(_row[0], 5)
self.assertEqual(_row[1], 50)
# Now test where there is no matching result:
_cursor.execute("SELECT dateTime, min FROM test1 WHERE dateTime = -1")
_row = _cursor.fetchone()
self.assertEqual(_row, None)
_cursor.close()
_connect.close()
开发者ID:Oga83,项目名称:weewx,代码行数:26,代码来源:test_weedb.py
示例4: test_no_tables
def test_no_tables(self):
weedb.create(self.db_dict)
_connect = weedb.connect(self.db_dict)
self.assertEqual(_connect.tables(), [])
self.assertRaises(weedb.ProgrammingError, _connect.columnsOf, 'test1')
self.assertRaises(weedb.ProgrammingError, _connect.columnsOf, 'foo')
_connect.close()
开发者ID:Oga83,项目名称:weewx,代码行数:7,代码来源:test_weedb.py
示例5: _init_db
def _init_db(stats_db_dict, stats_schema):
"""Create and initialize a database."""
# First, create the database if necessary. If it already exists, an
# exception will be thrown.
try:
weedb.create(stats_db_dict)
except weedb.DatabaseExists:
pass
# Get a connection
_connect = weedb.connect(stats_db_dict)
try:
# Now create all the necessary tables as one transaction:
with weedb.Transaction(_connect) as _cursor:
for _stats_tuple in stats_schema:
# Get the SQL string necessary to create the type:
_sql_create_str = _sql_create_string_factory(_stats_tuple)
_cursor.execute(_sql_create_str)
# Now create the meta table:
_cursor.execute(meta_create_str)
# Set the unit system to 'None' (Unknown) for now
_cursor.execute(meta_replace_str, ('unit_system', 'None'))
# Finally, save the stats schema:
StatsDb._save_schema(_cursor, stats_schema)
except Exception, e:
_connect.close()
syslog.syslog(syslog.LOG_ERR, "stats: Unable to create stats database.")
syslog.syslog(syslog.LOG_ERR, "**** %s" % (e,))
raise
开发者ID:HarleySchool,项目名称:commonssite,代码行数:31,代码来源:stats.py
示例6: open
def open(stats_db_dict):
"""Helper function to return an opened StatsDb object.
stats_db_dict: A dictionary passed on to weedb. It should hold
the keywords necessary to open the database."""
connection = weedb.connect(stats_db_dict)
return StatsDb(connection)
开发者ID:HarleySchool,项目名称:commonssite,代码行数:7,代码来源:stats.py
示例7: test_variable
def test_variable(self):
weedb.create(self.db_dict)
with weedb.connect(self.db_dict) as _connect:
_v = _connect.get_variable('lower_case_table_names')
self.assertTrue(_v[1] in ['0', '1', '2'], "Unknown lower_case_table_names value")
_v = _connect.get_variable('foo')
self.assertEqual(_v, None)
开发者ID:MLAB-project,项目名称:weewx,代码行数:7,代码来源:test_weedb.py
示例8: test_variable
def test_variable(self):
weedb.create(self.db_dict)
_connect = weedb.connect(self.db_dict)
_v = _connect.get_variable('journal_mode')
self.assertEqual(_v[1].lower(), 'delete')
_v = _connect.get_variable('foo')
self.assertEqual(_v, None)
_connect.close()
开发者ID:MiddleFork,项目名称:weewx,代码行数:8,代码来源:test_weedb.py
示例9: test
def test(db_dict):
weedb.create(db_dict)
connect = weedb.connect(db_dict)
cursor = connect.cursor()
cursor.execute("CREATE TABLE bar (col1 int, col2 int)")
with self.assertRaises(weedb.NoColumnError) as e:
cursor.execute("SELECT foo from bar")
cursor.close()
connect.close()
开发者ID:MLAB-project,项目名称:weewx,代码行数:9,代码来源:test_errors.py
示例10: test_select_nonexistent_database
def test_select_nonexistent_database(self):
mysql_dict = dict(mysql1_dict)
mysql_dict.pop('database_name')
connect = weedb.connect(mysql_dict)
cursor = connect.cursor()
with self.assertRaises(weedb.NoTableError):
cursor.execute("SELECT foo from test_weewx1.bar")
cursor.close()
connect.close()
开发者ID:MLAB-project,项目名称:weewx,代码行数:9,代码来源:test_errors.py
示例11: main
def main():
print "start time=", timestamp_to_string(start_ts)
print "stop time= ", timestamp_to_string(stop_ts)
print "(Approximately %.1f days of data)" % ((stop_ts - start_ts)/(24*3600.0))
print "***** SQLITE *****"
create_table(sqlite_db_dict)
connect = weedb.connect(sqlite_db_dict)
time_query(connect, 'outTemp')
time_query(connect, 'barometer')
connect.close()
print "***** MySQL *****"
create_table(mysql_db_dict)
connect = weedb.connect(mysql_db_dict)
time_query(connect, 'outTemp')
time_query(connect, 'barometer')
connect.close()
开发者ID:tkeffer,项目名称:benchmarks,代码行数:19,代码来源:time_db6nf.py
示例12: test_bad_select
def test_bad_select(self):
self.populate_db()
with weedb.connect(self.db_dict) as _connect:
with _connect.cursor() as _cursor:
# Test SELECT on a bad table name
self.assertRaises(weedb.ProgrammingError, _cursor.execute, "SELECT dateTime, min FROM foo")
# Test SELECT on a bad column name
self.assertRaises(weedb.OperationalError, _cursor.execute, "SELECT dateTime, foo FROM test1")
开发者ID:MLAB-project,项目名称:weewx,代码行数:10,代码来源:test_weedb.py
示例13: populate_db
def populate_db(self):
weedb.create(self.db_dict)
self.assertRaises(weedb.DatabaseExists, weedb.create, self.db_dict)
with weedb.connect(self.db_dict) as _connect:
with weedb.Transaction(_connect) as _cursor:
_cursor.execute("""CREATE TABLE test1 ( dateTime INTEGER NOT NULL UNIQUE PRIMARY KEY,
min REAL, mintime INTEGER, max REAL, maxtime INTEGER, sum REAL, count INTEGER, descript CHAR(20));""")
_cursor.execute("""CREATE TABLE test2 ( dateTime INTEGER NOT NULL UNIQUE PRIMARY KEY,
min REAL, mintime INTEGER, max REAL, maxtime INTEGER, sum REAL, count INTEGER, descript CHAR(20));""")
for irec in range(20):
_cursor.execute("INSERT INTO test1 (dateTime, min, mintime) VALUES (?, ?, ?)", (irec, 10*irec, irec))
开发者ID:MLAB-project,项目名称:weewx,代码行数:11,代码来源:test_weedb.py
示例14: test_rollback
def test_rollback(self):
# Create the database and schema
weedb.create(self.db_dict)
with weedb.connect(self.db_dict) as _connect:
with _connect.cursor() as _cursor:
_cursor.execute("""CREATE TABLE test1 ( dateTime INTEGER NOT NULL UNIQUE PRIMARY KEY, x REAL );""")
# Now start the transaction
_connect.begin()
for i in range(10):
_cursor.execute("""INSERT INTO test1 (dateTime, x) VALUES (?, ?)""", (i, i+1))
# Roll it back
_connect.rollback()
# Make sure nothing is in the database
with weedb.connect(self.db_dict) as _connect:
with _connect.cursor() as _cursor:
_cursor.execute("SELECT dateTime, x from test1")
_row = _cursor.fetchone()
self.assertEqual(_row, None)
开发者ID:MLAB-project,项目名称:weewx,代码行数:20,代码来源:test_weedb.py
示例15: open
def open(raw_db_dict):
"""Open a Raw database.
An exception of type weedb.OperationalError will be raised if the
database does not exist.
An exception of type StandardError will be raised if the database
exists, but has not been initialized.
Returns:
An instance of RawData."""
connection = weedb.connect(raw_db_dict)
return RawData(connection)
开发者ID:hoevenvd,项目名称:weewx_poller,代码行数:13,代码来源:raw.py
示例16: test_create
def test_create(self):
self.populate_db()
with weedb.connect(self.db_dict) as _connect:
self.assertEqual(sorted(_connect.tables()), ['test1', 'test2'])
self.assertEqual(_connect.columnsOf('test1'), ['dateTime', 'min', 'mintime', 'max', 'maxtime', 'sum', 'count', 'descript'])
self.assertEqual(_connect.columnsOf('test2'), ['dateTime', 'min', 'mintime', 'max', 'maxtime', 'sum', 'count', 'descript'])
for icol, col in enumerate(_connect.genSchemaOf('test1')):
self.assertEqual(schema[icol], col)
for icol, col in enumerate(_connect.genSchemaOf('test2')):
self.assertEqual(schema[icol], col)
# Make sure an IntegrityError gets raised in the case of a duplicate key:
with weedb.Transaction(_connect) as _cursor:
self.assertRaises(weedb.IntegrityError, _cursor.execute,
"INSERT INTO test1 (dateTime, min, mintime) VALUES (0, 10, 0)")
开发者ID:MLAB-project,项目名称:weewx,代码行数:14,代码来源:test_weedb.py
示例17: open
def open(archive_db_dict, table='archive'):
"""Open an Archive database.
An exception of type weedb.OperationalError will be raised if the
database does not exist.
An exception of type StandardError will be raised if the database
exists, but has not been initialized.
Returns:
An instance of Archive."""
_connect = weedb.connect(archive_db_dict)
return Archive(_connect, table)
开发者ID:crmorse,项目名称:weewx-waterflow,代码行数:14,代码来源:archive.py
示例18: test_bad_select
def test_bad_select(self):
self.populate_db()
_connect = weedb.connect(self.db_dict)
_cursor = _connect.cursor()
# Test SELECT on a bad table name
with self.assertRaises(weedb.ProgrammingError):
_cursor.execute("SELECT dateTime, min FROM foo")
# Test SELECT on a bad column name
with self.assertRaises(weedb.OperationalError):
_cursor.execute("SELECT dateTime, foo FROM test1")
_cursor.close()
_connect.close()
开发者ID:davidhrbaty,项目名称:weewx,代码行数:15,代码来源:test_weedb.py
示例19: create_table
def create_table(db_dict):
"""Create and populate the database table using a 6NF schema"""
# If the following is uncommented, the data will be deleted
# before every run.
# try:
# weedb.drop(db_dict)
# except weedb.NoDatabase:
# pass
# Try to create the database. If it already exists,
# an exception will be raised. Be prepared to catch it
try:
weedb.create(db_dict)
except weedb.DatabaseExists:
pass
connect = weedb.connect(db_dict)
cursor = connect.cursor()
# Create the table and generate the data. If it already exists,
# an exception will be raised. Be prepared to catch it
# and skip generating the data.
try:
# Note that every measurement gets its own row
# The primary key is the combination of the timestamp and observation type
cursor.execute("CREATE TABLE bench ("
"dateTime REAL NOT NULL, "
"obstype VARCHAR(63) NOT NULL, "
"measurement REAL, "
"CONSTRAINT pk PRIMARY KEY (dateTime, obstype))")
except weedb.OperationalError:
print "Benchmark data already exists"
else:
print "Generating fake data"
gen_data(connect)
finally:
cursor.close()
connect.close()
开发者ID:tkeffer,项目名称:benchmarks,代码行数:39,代码来源:time_db6nf.py
示例20: open_with_create
def open_with_create(raw_db_dict, rawSchema):
"""Open a Raw database, initializing it if necessary.
raw_db_dict: A database dictionary holding the information necessary
to open the database.
rawSchema: The schema to be used
Returns:
An instance of RawData"""
try:
rawData = RawData.open(raw_db_dict)
# The database exists and has been initialized. Return it.
return rawData
except (weedb.OperationalError, weewx.UninitializedDatabase):
pass
# First try to create the database. If it already exists, an exception will
# be thrown.
try:
weedb.create(raw_db_dict)
except weedb.DatabaseExists:
pass
# List comprehension of the types, joined together with commas. Put
# the SQL type in backquotes to avoid conflicts with reserved words
_sqltypestr = ', '.join(["`%s` %s" % _type for _type in rawSchema])
_connect = weedb.connect(raw_db_dict)
try:
with weedb.Transaction(_connect) as _cursor:
_cursor.execute("CREATE TABLE raw (%s);" % _sqltypestr)
except Exception, e:
_connect.close()
syslog.syslog(syslog.LOG_ERR, "raw: Unable to create database raw.")
syslog.syslog(syslog.LOG_ERR, "**** %s" % (e,))
raise
开发者ID:hoevenvd,项目名称:weewx_poller,代码行数:39,代码来源:raw.py
注:本文中的weedb.connect函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论