如何用sqlalchemy搜索数据库的存在

Question

我需要用python sqlalchemy编写一个脚本来搜索数据库是否存在，如果数据库存在，它应该查询数据库，否则创建数据库和表。

伪代码：

if db exists cursor.execute(sql) 
    else 
     create db test; 
     create tables; 
     insert data; 

Answer 1

创建连接到数据库，并执行类似SELECT 1;通用查询引擎。如果失败，您可以创建数据库。但是，如何创建新的数据库取决于DBMS。

使用PostgreSQL，您可以连接到postgres数据库并发出CREATE DATABASE语句，然后连接到新创建的数据库。

Answer 2

您可以使用sqlalchemy.engine.base.Engine.connect()方法，如果数据库不存在，将会产生OperationalError。

import sqlalchemy as sqla 
db = sqla.create_engine(database_uri) 
try: 
    db.connect() 
    db.execute(sql) 
except OperationalError: 
    # Switch database component of the uri 
    default_database_uri = os.path.join(os.path.dirname(
          str(db.engine.url)), 'mysql') 
    db = sqla.create_engine(default_database_uri) 
    # Create your missing database/tables/data here 
    # ... 

Answer 3

如果您不介意导入其他库，另一种方法是使用sqlalchemy_utils。然后database_exists做你期望的，你可以通过sqlalchemy数据库uri。

if database_exists('sqllite:////tmp/test.db'): 
    do_stuff_with_db(db) 

http://sqlalchemy-utils.readthedocs.org/en/latest/database_helpers.html

Answer 4

我不知道的正规途径是什么，但这里的检查，看是否通过对数据库列表中检查存在的数据库的方式。

from sqlalchemy import create_engine 

# This engine just used to query for list of databases 
mysql_engine = create_engine('mysql://{0}:{1}@{2}:{3}'.format(user, pass, host, port)) 

# Query for existing databases 
existing_databases = mysql_engine.execute("SHOW DATABASES;") 
# Results are a list of single item tuples, so unpack each tuple 
existing_databases = [d[0] for d in existing_databases] 

# Create database if not exists 
if database not in existing_databases: 
    mysql_engine.execute("CREATE DATABASE {0}".format(database)) 
    print("Created database {0}".format(database)) 

# Go ahead and use this engine 
db_engine = create_engine('mysql://{0}:{1}@{2}:{3}/{4}'.format(user, pass, host, port, db)) 

如果您不需要知道数据库是否已创建，那么这里有一种替代方法。

from sqlalchemy import create_engine 

# This engine just used to query for list of databases 
mysql_engine = create_engine('mysql://{0}:{1}@{2}:{3}'.format(user, pass, host, port)) 

# Query for existing databases 
mysql_engine.execute("CREATE DATABASE IF NOT EXISTS {0} ".format(database)) 

# Go ahead and use this engine 
db_engine = create_engine('mysql://{0}:{1}@{2}:{3}/{4}'.format(user, pass, host, port, db))