Sqlalchemy if table does not exist

Question

I wrote a module which is to create an empty database file

def create_database():
    engine = create_engine("sqlite:///myexample.db", echo=True)
    metadata = MetaData(engine)
    metadata.create_all()

But in another function, I want to open myexample.db database, and create tables to it if it doesn't already have that table.

EG of the first, subsequent table I would create would be:

Table(Variable_TableName, metadata,
       Column('Id', Integer, primary_key=True, nullable=False),
       Column('Date', Date),
       Column('Volume', Float))

(Since it is initially an empty database, it will have no tables in it, but subsequently, I can add more tables to it. Thats what i'm trying to say.)

Any suggestions?

Baneberry

What you are trying to do is database migrations. You should consider looking at Alembic. It's a database migration package for python to be used with SQLAlchemy. – Constellate

Answer 1

I've managed to figure out what I intended to do. I used engine.dialect.has_table(engine, Variable_tableName) to check if the database has the table inside. IF it doesn't, then it will proceed to create a table in the database.

Sample code:

engine = create_engine("sqlite:///myexample.db")  # Access the DB Engine
if not engine.dialect.has_table(engine, Variable_tableName):  # If table don't exist, Create.
    metadata = MetaData(engine)
    # Create a table with the appropriate Columns
    Table(Variable_tableName, metadata,
          Column('Id', Integer, primary_key=True, nullable=False), 
          Column('Date', Date), Column('Country', String),
          Column('Brand', String), Column('Price', Float),
    # Implement the creation
    metadata.create_all()

This seems to be giving me what i'm looking for.

Baneberry

i had to add schema = 'dbo' in has_table call to make it work on MS SQL – Usm

@Usm In what part of the code do you add "dbo"? Currently I have problems asking about a table in MSSQL. – Puffin

@vix I'm not sure I follow what you mean. Is sqlalchemy able to access the database you supplied it with? If not, it will not be able to query the table you are looking for. – Baneberry

@Puffin if not engine.dialect.has_table(engine, Variable_tableName, schema = 'dbo') – Usm

Using SQLAlchemy v1.2.11, engine.dialect.has_table(engine, tableName) is not there. However engine.has_table(tableName) does the job nicely :) – Tailspin

meta.create_all() has an implicit conditional, it creates the table if it does not exist. If that is the case we we need this "if" ? – Rillet

This is unnecessary as @Belle05's answer states, because create_all() has an optional parameter checkfirst that defaults to True. If set, tables are only created if they do not exist yet. – Unique

I get the following error when running your code: sqlalchemy.exc.ArgumentError: The argument passed to Dialect.has_table() should be a <class 'sqlalchemy.engine.base.Connection'>, got <class 'sqlalchemy.engine.base.Engine'>. How should I solve this? – Fatherly

Answer 2

Note that in 'Base.metadata' documentation it states about create_all:

Conditional by default, will not attempt to recreate tables already present in the target database.

And if you can see that create_all takes these arguments: create_all(self, bind=None, tables=None, checkfirst=True), and according to documentation:

Defaults to True, don't issue CREATEs for tables already present in the target database.

So if I understand your question correctly, you can just skip the condition.

Simonton

Answer 3

The accepted answer prints a warning that engine.dialect.has_table() is only for internal use and not part of the public API. The message suggests this as an alternative, which works for me:

import os
import sqlalchemy

# Set up a connection to a SQLite3 DB
test_db = os.getcwd() + "/test.sqlite"
db_connection_string = "sqlite:///" + test_db
engine = create_engine(db_connection_string)

# The recommended way to check for existence
sqlalchemy.inspect(engine).has_table("BOOKS")

---

See also the SQL Alchemy docs.

Convexity

Yes; this worked for me. [inspect] returns a Boolean, so very easy to use in all logical flows. I also see the Warning when I try to use [Dialect]. – Veinule

Answer 4

For those who define the table first in some models.table file, among other tables. This is a code snippet for finding the class that represents the table we want to create ( so later we can use the same code to just query it )

But together with the if written above, I still run the code with checkfirst=True

ORMTable.__table__.create(bind=engine, checkfirst=True)

models.table

class TableA(Base):
class TableB(Base):
class NewTableC(Base):

   id = Column('id', Text)
   name = Column('name', Text)

form

Then in the form action file:

engine = create_engine("sqlite:///myexample.db")
if not engine.dialect.has_table(engine, table_name):
   # Added to models.tables the new table I needed ( format Table as written above )
   table_models = importlib.import_module('models.tables')

   # Grab the class that represents the new table
   # table_name = 'NewTableC'
   ORMTable = getattr(table_models, table_name)            

   # checkfirst=True to make sure it doesn't exists
   ORMTable.__table__.create(bind=engine, checkfirst=True)

Neogothic

was looking everywhere for this; this worked for me – Stimulate

Answer 5

Tried that recently with SQLite. For some reason engine.dialect.has_table(engine, Variable_tableName) didn't work for me, however the following worked correctly:

engine = create_engine("sqlite:///myexample.db")
if engine.dialect.has_table(engine.connect(), Variable_tableName):
    ...

Lukewarm

Answer 6

engine.dialect.has_table does not work for me on cx_oracle. I am getting AttributeError: 'OracleDialect_cx_oracle' object has no attribute 'default_schema_name'

I wrote a workaround function:

from sqlalchemy.engine.base import Engine  
def orcl_tab_or_view_exists(in_engine: Engine, in_object: str,  in_object_name: str,)-> bool:
    """Checks if Oracle table exists in current in_engine connection

    in_object: 'table' | 'view'

    in_object_name: table_name | view_name
      """
    obj_query = """SELECT {o}_name FROM all_{o}s WHERE owner = SYS_CONTEXT ('userenv', 'current_schema') AND {o}_name = '{on}'
    """.format(o=in_object, on=in_object_name.upper())
    with in_engine.connect() as connection:
        result = connection.execute(obj_query)
    return len(list(result)) > 0

Sterilize

Answer 7

This is the code working for me to create all tables of all model classes defined with Base class

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

class YourTable(Base):
    __tablename__ = 'your_table'
    id            = Column(Integer, primary_key = True)

DB_URL="mysql+mysqldb://<user>:<password>@<host>:<port>/<db_name>"
scoped_engine = create_engine(DB_URL)
Base = declarative_base()
Base.metadata.create_all(scoped_engine)

Freedwoman

Answer 8

Just place this line at the end of your model classes, (note: replace BaseModel name with your Base model class name) to auto-generate all database tables for each model given.

# create tables
BaseModel.metadata.create_all(bind=engine)

Full sample code given below.

import sqlite3
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

SQLALCHEMY_DATABASE_URL = 'sqlite:///./assets/db/ert_0.db'

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False}, echo=True)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

BaseModel = declarative_base()

# sample model - bind with app_state table in the db
class AppState(BaseModel):
    __tablename__ = 'app_state'

    id = Column(Integer, primary_key=True, index=True)
    key = Column(String(255), unique=True, index=True)
    value = Column(String(512))

# create tables - place this line after all the models 
BaseModel.metadata.create_all(bind=engine)

note: SQLAlchemy version 2.0.31

Xe