Как создать новую базу данных с помощью SQLAlchemy?

Используя SQLAlchemy, объект Engine создается следующим образом:

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

Доступ engineне удается, если база данных, указанная в аргументе create_engine(в данном случае mydb), не существует. Можно ли указать SQLAlchemy создать новую базу данных, если указанная база данных не существует?

В postgres по умолчанию обычно присутствуют три базы данных. Если вы можете подключиться как суперпользователь (например, с postgresролью), вы можете подключиться к базам данных postgresили template1. По умолчанию pg_hba.conf разрешает postgresиспользовать postgresроль только указанному пользователю unix , поэтому проще всего просто стать этим пользователем. В любом случае создайте движок как обычно с пользователем, у которого есть разрешения на создание базы данных:

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

Однако вы не можете использовать engine.execute(), потому что postgres не позволяет создавать базы данных внутри транзакций, а sqlalchemy всегда пытается запускать запросы в транзакции. Чтобы обойти это, получите базовое соединение от движка:

>>> conn = engine.connect()

Но соединение по-прежнему будет внутри транзакции, поэтому вы должны завершить открытую транзакцию с помощью commit:

>>> conn.execute("commit")

Затем вы можете приступить к созданию базы данных, используя для этого соответствующую команду PostgreSQL.

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

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

Это позволяет избежать ручного управления транзакциями при создании базы данных, предоставляя 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

Обратите внимание, что я не мог получить вышеуказанные предложения, database_existsпотому что всякий раз, когда я проверяю, существует ли база данных, используя if not, database_exists(engine.url):я получаю эту ошибку:

InterfaceError ('(pyodbc.InterfaceError) (\' 28000 \ ', u \' [28000] [Microsoft] [Собственный клиент SQL Server 11.0] [SQL Server]) Ошибка входа для пользователя \\ 'myUser \\'. (18456) (SQLDriverConnect); [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] Невозможно открыть базу данных «MY_DATABASE», запрошенную при входе в систему. Ошибка входа. (4060); [28000] [Microsoft] [SQL Server Native Клиент 11.0] [SQL Server] Ошибка входа для пользователя \\ 'myUser \\'. (18456); [28000] [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))