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

Question

С помощью SQLAlchemy движок создается так:

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

Доступ к engine не удается, если база данных отсутствует. Можно ли указать SQLAlchemy создать новую базу данных, если указанная база данных не существует?

Answer 1

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

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

Answer 2

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

Answer 3

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

Answer 4

Расширение принятого ответа с помощью with дает:

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

NEW_DB_NAME = 'database_name'

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

Answer 5

Обратите внимание, что я не смог получить приведенные выше предложения с 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))  

Answer 6

Если кто-то вроде меня не хочет использовать весь sqlalchemy_utils в своем проекте только для создания базы данных, вы можете использовать такой скрипт. Я пришел с этим, основываясь на ответе SingleNegationElimination.Я использую здесь 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")