SQL 화학을 사용하여 새 데이터베이스를 만드는 방법은 무엇입니까?

SQL 화학을 사용하여 새 데이터베이스를 만드는 방법은 무엇입니까?

SQLAlchemy를 사용하면 다음과 같이 Engine 개체가 생성됩니다.

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

액세스engine인수에 지정된 데이터베이스가 실패할 경우create_engine(이 경우에는,mydb)가 존재하지 않습니다.지정된 데이터베이스가 존재하지 않는 경우 SQL 화학에서 새 데이터베이스를 생성하도록 지시할 수 있습니까?

SQLAlchemy-Utils는 SQLAlchemy에 대한 사용자 지정 데이터 유형 및 다양한 유틸리티 기능을 제공합니다.pip을 사용하여 최신 공식 버전을 설치할 수 있습니다.

pip install sqlalchemy-utils

데이터베이스 도우미에는 다음이 포함됩니다.create_database함수:

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

postgres에서는 일반적으로 세 개의 데이터베이스가 기본적으로 존재합니다.슈퍼유저로 연결할 수 있는 경우(예:postgres역할), 그런 다음 에 연결할 수 있습니다.postgres또는template1데이터베이스.기본 pg_hba.conf는 이름이 지정된 unix 사용자만 허용합니다.postgres를 사용하기 위해postgres역할, 즉 가장 간단한 것은 그 사용자가 되는 것입니다.어쨌든 데이터베이스를 만들 수 있는 권한을 가진 사용자가 있는 경우에는 평소와 같이 엔진을 만듭니다.

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

사용할 수 없습니다.engine.execute()그러나 postgres에서는 트랜잭션 내부에 데이터베이스를 만들 수 없으므로 sqlalchemy에서는 항상 트랜잭션에서 쿼리를 실행하려고 합니다.이 문제를 해결하려면 엔진에서 기본 연결을 가져옵니다.

>>> conn = engine.connect()

그러나 연결은 여전히 트랜잭션 내부에 있으므로 열린 트랜잭션을 종료해야 합니다.commit:

>>> conn.execute("commit")

그런 다음 적절한 Postgre를 사용하여 데이터베이스를 작성할 수 있습니다.SQL 명령을 사용합니다.

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

다음을 제공하여 데이터베이스를 생성하는 동안 수동 트랜잭션 관리를 피할 수 있습니다.isolation_level='AUTOCOMMIT'로.create_engine함수:

import sqlalchemy

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

또한 데이터베이스가 존재하지 않는지 확신할 수 없는 경우 를 억제하여 존재로 인한 데이터베이스 생성 오류를 무시할 수 있습니다.sqlalchemy.exc.ProgrammingError예외:

import contextlib
import sqlalchemy.exc

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

다음을 사용하여 승인된 답변 확장with산출량:

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

위와 같은 제안을 받을 수 없음을 참고하시기 바랍니다.database_exists왜냐하면 내가 데이터베이스가 존재하는지 확인할 때마다 그렇지 않으면 사용하기 때문입니다.database_exists(engine.url):다음 오류가 발생했습니다.

인터페이스 오류('(pyodbc).인터페이스 오류)(\'28000\', u\'[28000] [Microsoft][SQL Server Native Client 11.0][SQL 서버]사용자 \\'myUser\".(18456)(SQLDriverConnect); [28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]로그인에서 요청한 데이터베이스 "MY_DATABASE"를 열 수 없습니다.로그인에 실패했습니다. (4060); [28000] [Microsoft] [SQL Server Native Client 11.0][SQL 서버]사용자 \\'myUser\"(18456), [28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]로그인에서 요청한 데이터베이스 "MY_DATABASE"를 열 수 없습니다.로그인에 실패했습니다. (4060)\',)

도.contextlib/suppress작동하지도 않았고 사용하지도 않았습니다.postgres데이터베이스가 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))  

저와 같은 사람이 데이터베이스 생성을 위해 전체 sqlalchemy_utils를 프로젝트에 가져가고 싶지 않다면, 다음과 같은 스크립트를 사용할 수 있습니다.싱글 네거티브 엘리미네이션의 답변에 따라 함께 왔습니다.여기서 pydantic(FastAPI 프로젝트)과 가져온 설정을 참조용으로 사용하고 있지만 쉽게 변경할 수 있습니다.

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

언급URL : https://stackoverflow.com/questions/6506578/how-to-create-a-new-database-using-sqlalchemy