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

【问题标题】：how to search for the existence of a database with sqlalchemy如何使用 sqlalchemy 搜索数据库的存在
【发布时间】：2021-09-19 11:43:58
【问题描述】：

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

伪代码：

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

【问题讨论】：

标签： sqlalchemy

【解决方案1】：

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

import sqlalchemy as sqla
from sqlalchemy import create_engine
from sqlalchemy.exc import OperationalError
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
    # ...

【讨论】：

connect() 不会引发异常，而是创建并清空一个异常

【解决方案2】：

如果您不介意导入其他库，另一种方法是使用 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

【讨论】：

这种方法在 Python 3.8 中似乎还行不通
它适用于我的 Python 3.8，但如果 SQLite DB 位于 /tmp 下的某个位置，不

【解决方案3】：

我不知道规范方法是什么，但这里有一种方法可以通过检查数据库列表来检查数据库是否存在。

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))

【讨论】：

【解决方案4】：

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

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

【讨论】：

我已经存在一个数据库池，所以如果我创建一个通用查询，它将选择那些而不是创建我想要的那个。如果不存在创建数据库查询，我希望有一个相当于 mysql 的 sqlalchemy。
“通用”是指它适用于所有 SQL 数据库——我不知道您使用的是哪个后端。 create_engine(my_target_dsn).execute('select 1') 不会干扰您现有的游泳池。
我的错，我明白了，我想到了这一点，但它看起来很简单，我认为它不会起作用。顺便说一句，我正在使用 python 和 sqlalchemy

【解决方案5】：

您可以使用https://sqlalchemy-utils.readthedocs.io/en/latest/ 有一个助手叫database_exists()

【讨论】：

【解决方案6】：

如果您决定使用 sqlalchemy_utils，请确保将您的 connect_args 作为查询字符串应用。否则当database_exists 函数recreates the sqlalchemy engine 时，这些参数不会被结转

from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists

connect_args = {'charset': 'utf8'}
connect_args['ssl_cert'] = 'certs/client-cert.pem'
connect_args['ssl_key'] = 'certs/client-key.pem'
connect_args['ssl_ca'] = 'certs/server-ca.pem'

engine = create_engine(os.environ['MYSQL_CONN_URL'] + '/' + os.environ['DB_NAME'] + '?' +  urlencode(connect_args))
print(database_exists(engine.url))

【讨论】：

【解决方案7】：

安装 pip 包

pip install sqlalchemy_utils

检查 database_exists

from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists

engine = create_engine('postgresql://postgres@localhost/name')
database_exists(engine.url)

【讨论】：