如何使用 SQLAlchemy 创建新数据库？

通过使用 SQLAlchemy，可以像下面这样创建 Engine 对象:

from sqlalchemy import create_engine
engine = create_engine("postgresql://localhost/mydb")

如果 create_engine参数中指定的数据库(在本例中为 mydb)不存在，则访问 engine失败。如果指定的数据库不存在，是否可以告诉 SQLAlchemy 创建一个新数据库？

126763 次浏览
小开
最佳答案

在 postgres 上，默认情况下通常存在三个数据库。如果您能够以超级用户的身份连接(例如，postgres角色) ，那么您可以连接到 postgres或 template1数据库。默认的 pg _ hba。Conf 只允许名为 postgres的 unix 用户使用 postgres角色，因此最简单的方法就是成为该用户。无论如何，与通常一样，创建一个拥有创建数据库权限的用户的引擎:

>>> engine = sqlalchemy.create_engine("postgres://postgres@/postgres")

但是，您不能使用 engine.execute()，因为 postgres 不允许您在事务中创建数据库，并且 sqltancy 总是尝试在事务中运行查询。要解决这个问题，从引擎获取底层连接:

>>> conn = engine.connect()

但是连接仍然在一个事务中，因此您必须以 commit结束打开的事务:

>>> conn.execute("commit")

然后，您可以继续使用适当的 PostgreSQL 命令创建数据库。

>>> conn.execute("create database test")
>>> conn.close()
小开

通过提供 isolation_level='AUTOCOMMIT'到 create_engine函数，在创建数据库时可以避免手动事务管理:

import sqlalchemy


with sqlalchemy.create_engine(
'postgresql:///postgres',
isolation_level='AUTOCOMMIT'
).connect() as connection:
connection.execute('CREATE DATABASE my_database')

另外，如果您不确定数据库是否存在，可以通过抑制 sqlalchemy.exc.ProgrammingError异常来忽略由于存在而导致的数据库创建错误:

import contextlib
import sqlalchemy.exc


with contextlib.suppress(sqlalchemy.exc.ProgrammingError):
# creating database as above
小开

SQLAlchemy-Utils 为 SQLAlchemy 提供自定义数据类型和各种实用函数。您可以使用 pip 安装最新的官方版本:

pip install sqlalchemy-utils

资料库辅助工具包括 create_database功能:

from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database


engine = create_engine("postgres://localhost/mydb")
if not database_exists(engine.url):
create_database(engine.url)


print(database_exists(engine.url))
小开

请注意，我不能得到上述建议与 database_exists，因为每当我检查数据库是否存在，如果没有 database_exists(engine.url):我得到这个错误:

InterfaceError (’(pyodbc.InterfaceError)(’28000’，u’[28000] [ Microsoft ][ SQL Server 本机客户端11.0][ SQL Server ]登录失败 User‘ myUser’(18456)(SQLDriverConnect) ; [28000] [ Microsoft ][ SQL Server 本机客户端11.0][ SQL Server ]无法打开 登录请求的数据库“ MY _ DATABASE”。登录失败。 (4060) ; [28000][ Microsoft ][ SQL Server 本机客户端11.0][ SQL 服务器]用户‘ myUser’登录失败 [ Microsoft ][ SQL Server 本机客户端11.0][ SQL Server ]无法打开 登录请求的数据库“ MY _ DATABASE”。登录失败。 (4060)’)’,)

另外，contextlib/suppress也不能正常工作，我也没有使用 postgres，所以我最后这样做是为了忽略这个异常，如果数据库碰巧已经存在于 SQL Server 中:

import logging
import sqlalchemy


logging.basicConfig(filename='app.log', format='%(asctime)s-%(levelname)s-%(message)s', level=logging.DEBUG)
engine = create_engine('mssql+pyodbc://myUser:mypwd@localhost:1234/MY_DATABASE?driver=SQL+Server+Native+Client+11.0?trusted_connection=yes', isolation_level = "AUTOCOMMIT")


try:
engine.execute('CREATE DATABASE ' + a_database_name)
except Exception as db_exc:
logging.exception("Exception creating database: " + str(db_exc))
小开

利用 with扩大公认答案的收益率:

from sqlalchemy import create_engine
engine = create_engine("postgresql://localhost")


NEW_DB_NAME = 'database_name'


with engine.connect() as conn:
conn.execute("commit")
# Do not substitute user-supplied database names here.
conn.execute(f"CREATE DATABASE {NEW_DB_NAME}")
小开

如果像我这样的人不想仅仅为了创建数据库而将整个 sqlchemy _ utils 带到您的项目中，您可以使用这样的脚本。我是根据 消除单一否定的答案来的。我在这里使用 pydantic (它是 FastAPI 项目)和我导入的参考设置，但是你可以很容易地改变这一点:

from sqlalchemy import create_engine
from sqlalchemy.exc import OperationalError
from pydantic import PostgresDsn


from src.conf import settings




def build_db_connection_url(custom_db: Optional[str] = None):
db_name = f"/{settings.POSTGRES_DB or ''}" if custom_db is None else "/" + custom_db
return PostgresDsn.build(
scheme='postgresql+psycopg2',
user=settings.POSTGRES_USER,
password=settings.POSTGRES_PASSWORD,
host=settings.POSTGRES_HOST,
path=db_name,
)




def create_database(db_name: str):
try:
eng = create_engine(build_db_connection_url(custom_db=db_name))
conn = eng.connect()
conn.close()
except OperationalError as exc:
if "does not exist" in exc.__str__():
eng = create_engine(build_db_connection_url(custom_db="postgres"))
conn = eng.connect()
conn.execute("commit")
conn.execute(f"create database {db_name}")
conn.close()
print(f"Database {db_name} created")
else:
raise exc
eng.dispose()


create_database("test_database")

提交答案