Skip to main content
deleted 8 characters in body
Source Link
Dmitry
  • 587
  • 8
  • 9

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()
        del eng
    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()
            # TODO: log about db creation
            print(f"Database {db_name} created")
        else:
            raise exc
    eng.dispose()

create_database("test_database")

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


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()
        del eng
    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()
            # TODO: log about db creation
            print(f"Database {db_name} created")
        else:
            raise exc

create_database("test_database")

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")
Source Link
Dmitry
  • 587
  • 8
  • 9

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


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()
        del eng
    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()
            # TODO: log about db creation
            print(f"Database {db_name} created")
        else:
            raise exc

create_database("test_database")