¿Cómo crear una nueva base de datos usando SQLAlchemy?

Usando SQLAlchemy, un objeto Engine se crea así:

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

El acceso enginefalla si la base de datos especificada en el argumento a create_engine(en este caso, mydb) no existe. ¿Es posible decirle a SQLAlchemy que cree una nueva base de datos si la base de datos especificada no existe?

En postgres, normalmente hay tres bases de datos presentes de forma predeterminada. Si puede conectarse como superusuario (por ejemplo, el postgresrol), puede conectarse a las bases de datos postgreso template1. El pg_hba.conf predeterminado permite que solo el usuario de Unix nombrado postgresuse el postgresrol, por lo que lo más simple es convertirse en ese usuario. En cualquier caso, cree un motor como de costumbre con un usuario que tenga los permisos para crear una base de datos:

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

engine.execute()Sin embargo, no puede usarlo , porque postgres no le permite crear bases de datos dentro de las transacciones y sqlalchemy siempre intenta ejecutar consultas en una transacción. Para evitar esto, obtenga la conexión subyacente del motor:

>>> conn = engine.connect()

Pero la conexión seguirá estando dentro de una transacción, por lo que debe finalizar la transacción abierta con commit:

>>> conn.execute("commit")

Y luego puede proceder a crear la base de datos utilizando el comando PostgreSQL adecuado para ello.

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

SQLAlchemy-Utils proporciona tipos de datos personalizados y varias funciones de utilidad para SQLAlchemy. Puede instalar la versión oficial más reciente usando pip:

pip install sqlalchemy-utils

Los ayudantes de la base de datos incluyen una create_databasefunción:

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

Es posible evitar la gestión manual de transacciones mientras se crea una base de datos proporcionando isolation_level='AUTOCOMMIT'a la create_enginefunción:

import sqlalchemy

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

Además, si no está seguro de que la base de datos no existe, hay una forma de ignorar el error de creación de la base de datos debido a la existencia suprimiendo la sqlalchemy.exc.ProgrammingErrorexcepción:

import contextlib
import sqlalchemy.exc

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

Tenga en cuenta que no pude obtener las sugerencias anteriores database_existsporque cada vez que verifico si la base de datos existe usando si no database_exists(engine.url):, aparece este error:

InterfaceError ('(pyodbc.InterfaceError) (\' 28000 \ ', u \' [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] Error al iniciar sesión para el usuario \\ 'myUser \\'. (18456) (SQLDriverConnect); [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] No se puede abrir la base de datos "MY_DATABASE" solicitada por el inicio de sesión. El inicio de sesión falló. (4060); [28000] [Microsoft] [SQL Server Native Cliente 11.0] [SQL Server] Error al iniciar sesión para el usuario \\ 'myUser \\'. (18456); [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] No se puede abrir la base de datos "MY_DATABASE" solicitada por el inicio de sesión . El inicio de sesión falló. (4060) \ ')',)

Además contextlib/suppress, no estaba funcionando y no estoy usando, postgresasí que terminé haciendo esto para ignorar la excepción si la base de datos ya existe con 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))