How to create a new database using SQLAlchemy?

user_189776 0
Viewed 121198

Using SQLAlchemy, an Engine object is created like this:

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

Accessing engine fails if the database specified in the argument to create_engine (in this case, mydb) does not exist. Is it possible to tell SQLAlchemy to create a new database if the specified database doesn't exist?

sqlalchemy python
6 Answers

Extending the accepted answer using with yields:

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

NEW_DB_NAME = 'database_name'

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

Please note that I couldn't get the above suggestions with database_exists because whenever I check if the database exists using if not database_exists(engine.url): I get this error:

InterfaceError('(pyodbc.InterfaceError) (\'28000\', u\'[28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]Login failed for user \\'myUser\\'. (18456) (SQLDriverConnect); [28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]Cannot open database "MY_DATABASE" requested by the login. The login failed. (4060); [28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]Login failed for user \\'myUser\\'. (18456); [28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]Cannot open database "MY_DATABASE" requested by the login. The login failed. (4060)\')',)

Also contextlib/suppress was not working and I'm not using postgres so I ended up doing this to ignore the exception if the database happens to already exist with 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))
user_3511540

If someone like me don't want to take whole sqlalchemy_utils to your project just for database creation, you can use script like this. I've come with it, based on SingleNegationElimination's answer. I'm using pydantic here (it's FastAPI project) and my imported settings for reference, but you can easily change this:

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")
user_85377390
Related