I am using SQLAlchemy to generate tables in a specific schema in a PostgreSQL database. If the schema does not exist, I want to create it. I know the PostgreSQL query to check for the existence of the schema:
SELECT exists(select schema_name FROM information_schema.schemata WHERE schema_name = 'foo')
but I want to know how I should handle this using SQLAlchemy.
@javax's answer is almost correct; the following is a little clarification:
q = exists(select([("schema_name")]).select_from("information_schema.schemata")
.where("schema_name = 'foo'"))
if not session.query(q).scalar():
session.execute('CREATE SCHEMA foo;')
Textual column expression 'schema_name' should be explicitly declared with text('schema_name'), or use column('schema_name') for more specificity (this warning may be suppressed after 10 occurrences). To avoid just wrap with text() each string. ie: q = exists(select([(text("schema_name"))]).select_from(text("information_schema.schemata")).where(text("schema_name = 'foo'")))If you want to integrate it with SQLAlchemy you could use reflection but for an easier and quicker solution:
from sqlalchemy.sql import exists, select
exists(select([("schema_name")]).select_from("information_schema.schemata").
where("schema_name == 'foo'"))
This will return True or False.
I'm using MySQL and this works for me (with sqlalchemy v1.4.1, python 3.9.1):
import sqlalchemy as sa
engine = sa.create_engine("mysql+pymysql://someuser:somepassword@somehost")
inspector = sa.inspect(engine)
myschema = "someschema"
if myschema in inspector.get_schema_names():
print(f"{myschema} schema exists")
and if you want to create a schema if it doesn't exist:
if myschema not in inspector.get_schema_names():
engine.execute(sa.schema.CreateSchema(myschema))
# optional. set the default schema to the new schema:
engine.dialect.default_schema_name = myschema
I've been using this with Postgres, though was surprised to learn that IF NOT EXISTS is not part of the SQL standard -
engine.execute('CREATE SCHEMA IF NOT EXISTS foo;')
Apparently it's an extension for Postgres and MySQL - https://www.w3resource.com/sql/sql-basic/create-schema.php
This worked for me. Finds if a schema exists using pure SqlAlchemy:
from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy.schema import CreateSchema, DropSchema
...
engine = create_engine(dbstr)
meta = ModelBase()
...
if not engine.dialect.has_schema(engine, schema=schema_name):
event.listen(meta.metadata, 'before_create', CreateSchema(schema_name))
meta.metadata.reflect(bind=engine, schema=schema_name)
meta.metadata.create_all(engine, checkfirst=True)
SQLAlchemy 2.0 introduced backend-agnostic way to inspect the database state with Inspector object. It has has_schema method to check if schema exists in database.
I usually use this code if I need to create a new schema if it hasn't been yet presented in the database:
from sqlalchemy import create_engine, inspect
from sqlalchemy.schema import CreateSchema
schema = "new_schema"
connectable = create_engine("postgresql+psycopg2://postgres@localhost/postgres")
with connectable.connect() as connection:
if not inspect(connection).has_schema(schema):
connection.execute(CreateSchema(schema))
connection.commit()
