Como criar um novo banco de dados usando SQLAlchemy?

Usando SQLAlchemy, um objeto Engine é criado assim:

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

O acesso enginefalhará se o banco de dados especificado no argumento para create_engine(neste caso, mydb) não existir. É possível dizer ao SQLAlchemy para criar um novo banco de dados se o banco de dados especificado não existir?

No postgres, três bancos de dados normalmente estão presentes por padrão. Se você puder se conectar como um superusuário (por exemplo, a postgresfunção), poderá se conectar aos bancos de dados postgresou template1. O pg_hba.conf padrão permite que apenas o usuário Unix nomeado postgresuse a postgresfunção, então a coisa mais simples é simplesmente se tornar esse usuário. De qualquer forma, crie um mecanismo como de costume com um usuário que tenha as permissões para criar um banco de dados:

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

engine.execute()No entanto, você não pode usar , porque o postgres não permite que você crie bancos de dados dentro de transações e o sqlalchemy sempre tenta executar consultas em uma transação. Para contornar isso, obtenha a conexão subjacente do mecanismo:

>>> conn = engine.connect()

Mas a conexão ainda estará dentro de uma transação, então você deve encerrar a transação aberta com commit:

>>> conn.execute("commit")

E você pode então prosseguir para criar o banco de dados usando o comando PostgreSQL apropriado para ele.

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

SQLAlchemy-Utils fornece tipos de dados personalizados e várias funções de utilitário para SQLAlchemy. Você pode instalar a versão oficial mais recente usando pip:

pip install sqlalchemy-utils

Os auxiliares de banco de dados incluem uma create_databasefunção:

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

É possível evitar o gerenciamento manual de transações durante a criação do banco de dados fornecendo isolation_level='AUTOCOMMIT'para create_enginefuncionar:

import sqlalchemy

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

Além disso, se você não tiver certeza de que o banco de dados não existe, há uma maneira de ignorar o erro de criação do banco de dados devido à existência, suprimindo a sqlalchemy.exc.ProgrammingErrorexceção:

import contextlib
import sqlalchemy.exc

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

Observe que não consegui obter as sugestões acima com database_existsporque sempre que verifico se o banco de dados existe usando senão database_exists(engine.url):, recebo este erro:

InterfaceError ('(pyodbc.InterfaceError) (\' 28000 \ ', u \' [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] Falha de logon do usuário \\ 'myUser \\'. (18456) (SQLDriverConnect); [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] Não é possível abrir o banco de dados "MY_DATABASE" solicitado pelo login. O login falhou. (4060); [28000] [Microsoft] [SQL Server Native Cliente 11.0] [SQL Server] Falha de logon do usuário \\ 'meuUsuário \\'. (18456); [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] Não é possível abrir o banco de dados "MY_DATABASE" solicitado pelo logon . O login falhou. (4060) \ ')',)

Também contextlib/suppressnão estava funcionando e não estou usando, postgresentão acabei fazendo isso para ignorar a exceção caso o banco de dados já exista com o 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))