Jak stworzyć nową bazę danych za pomocą SQLAlchemy?

Za pomocą SQLAlchemy tworzony jest obiekt Engine w następujący sposób:

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

Dostęp enginenie powiedzie się, jeśli baza danych określona w argumencie do create_engine(w tym przypadku mydb) nie istnieje. Czy można powiedzieć SQLAlchemy, aby utworzył nową bazę danych, jeśli określona baza danych nie istnieje?

W Postgres domyślnie są obecne trzy bazy danych. Jeśli możesz połączyć się jako superużytkownik (np. postgresRola), możesz połączyć się z bazami danych postgreslub template1. Domyślny pg_hba.conf zezwala tylko unixowemu użytkownikowi nazwanemu postgresna używanie postgresroli, więc najprościej jest po prostu zostać tym użytkownikiem. W każdym razie utwórz silnik w zwykły sposób z użytkownikiem, który ma uprawnienia do tworzenia bazy danych:

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

Nie możesz jednak użyć engine.execute(), ponieważ postgres nie pozwala na tworzenie baz danych wewnątrz transakcji, a sqlalchemy zawsze próbuje uruchamiać zapytania w transakcji. Aby obejść ten problem, uzyskaj podstawowe połączenie z silnika:

>>> conn = engine.connect()

Ale połączenie nadal będzie wewnątrz transakcji, więc musisz zakończyć otwartą transakcję za pomocą commit:

>>> conn.execute("commit")

Następnie możesz przystąpić do tworzenia bazy danych za pomocą odpowiedniego polecenia PostgreSQL.

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

SQLAlchemy-Utils zapewnia niestandardowe typy danych i różne funkcje narzędziowe dla SQLAlchemy. Możesz zainstalować najnowszą oficjalną wersję za pomocą pip:

pip install sqlalchemy-utils

W pomocnicy baza zawiera create_databasefunkcję:

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

Podczas tworzenia bazy danych można uniknąć ręcznego zarządzania transakcjami, udostępniając isolation_level='AUTOCOMMIT'do create_enginefunkcji:

import sqlalchemy

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

Ponadto, jeśli nie jesteś pewien, że baza danych nie istnieje, istnieje sposób na zignorowanie błędu tworzenia bazy danych z powodu istnienia poprzez sqlalchemy.exc.ProgrammingErrorpominięcie wyjątku:

import contextlib
import sqlalchemy.exc

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

Zwróć uwagę, że nie mogłem uzyskać powyższych sugestii, database_existsponieważ za każdym razem, gdy sprawdzam, czy baza danych istnieje, używając polecenia Jeśli nie database_exists(engine.url):, otrzymuję ten błąd:

InterfaceError ('(pyodbc.InterfaceError) (\' 28000 \ ', u \' [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] Logowanie nie powiodło się dla użytkownika \\ 'myUser \\'. (18456) (SQLDriverConnect); [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] Nie można otworzyć bazy danych „MY_DATABASE” żądanej przy logowaniu. Logowanie nie powiodło się. (4060); [28000] [Microsoft] [SQL Server Native Klient 11.0] [SQL Server] Logowanie nie powiodło się dla użytkownika \\ 'myUser \\'. (18456); [28000] [Microsoft] [SQL Server Native Client 11.0] [SQL Server] Nie można otworzyć bazy danych „MY_DATABASE” żądanej przy logowaniu . Logowanie nie powiodło się. (4060) \ ')',)

Również contextlib/suppressnie działał i nie używam, postgreswięc w końcu zrobiłem to, aby zignorować wyjątek, jeśli baza danych już istnieje w 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))