psycopg2.errors.UndefinedObject: type MY_ENUM does not exist

[br-follow]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import pytest
from pytest_postgresql.compat import connection
from sqlmodel import Session, SQLModel, create_engine


@pytest.fixture(name="session")
def session_fixture(postgresql: connection):
    engine = create_engine(
        "postgresql+psycopg2://",
        connect_args=postgresql.get_dsn_parameters(),
        pool_pre_ping=True,
    )

    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session



----------------

SQLModel

class TaxIdType(enum.Enum):
    USA_SSN = "USA_SSN"
    NOT_SPECIFIED = "NOT_SPECIFIED"

class IdentityBase(SQLModel):
    pass

class IdentityCreate(IdentityBase):
    given_name: str = Field(..., description=", e.g., John")
    ...
    tax_id_type: Optional[TaxIdType] = Field(None, sa_column=Column(Enum(TaxIdType)))

Description

• Create a pytest fixture using pytest_postgresql to create a connection to a postgres db and return a session
• Error thrown when calling SQLModel.metadata.create_all(engine)

________________________________________________ ERROR at setup of test_createUser_succeeds _________________________________________________

self = <sqlalchemy.future.engine.Connection object at 0x1044ce430>
dialect = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x1044cebe0>
constructor = <bound method DefaultExecutionContext._init_ddl of <class 'sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2'>>
statement = '\nCREATE TABLE identity (\n\ttax_id_type taxidtype, \n\tfunding_source fundingsource[], \n\tgiven_name VARCHAR NOT NU...T, \n\tid UUID NOT NULL, \n\tuser_id UUID, \n\tPRIMARY KEY (id), \n\tFOREIGN KEY(user_id) REFERENCES users (id)\n)\n\n'
parameters = {}, execution_options = immutabledict({'autocommit': True})
args = (<sqlalchemy.dialects.postgresql.base.PGDDLCompiler object at 0x10452e2b0>,), kw = {}
branched = <sqlalchemy.future.engine.Connection object at 0x1044ce430>, conn = <sqlalchemy.pool.base._ConnectionFairy object at 0x1046ae160>
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x10452e2e0>
cursor = <cursor object at 0x1043069a0; closed: -1>, evt_handled = False

    def _execute_context(
        self,
        dialect,
        constructor,
        statement,
        parameters,
        execution_options,
        *args,
        **kw
    ):
        """Create an :class:`.ExecutionContext` and execute, returning
        a :class:`_engine.CursorResult`."""

        branched = self
        if self.__branch_from:
            # if this is a "branched" connection, do everything in terms
            # of the "root" connection, *except* for .close(), which is
            # the only feature that branching provides
            self = self.__branch_from

        try:
            conn = self._dbapi_connection
            if conn is None:
                conn = self._revalidate_connection()

            context = constructor(
                dialect, self, conn, execution_options, *args, **kw
            )
        except (exc.PendingRollbackError, exc.ResourceClosedError):
            raise
        except BaseException as e:
            self._handle_dbapi_exception(
                e, util.text_type(statement), parameters, None, None
            )

        if (
            self._transaction
            and not self._transaction.is_active
            or (
                self._nested_transaction
                and not self._nested_transaction.is_active
            )
        ):
            self._invalid_transaction()

        elif self._trans_context_manager:
            TransactionalContext._trans_ctx_check(self)

        if self._is_future and self._transaction is None:
            self._autobegin()

        context.pre_exec()

        if dialect.use_setinputsizes:
            context._set_input_sizes()

        cursor, statement, parameters = (
            context.cursor,
            context.statement,
            context.parameters,
        )

        if not context.executemany:
            parameters = parameters[0]

        if self._has_events or self.engine._has_events:
            for fn in self.dispatch.before_cursor_execute:
                statement, parameters = fn(
                    self,
                    cursor,
                    statement,
                    parameters,
                    context,
                    context.executemany,
                )

        if self._echo:

            self._log_info(statement)

            stats = context._get_cache_stats()

            if not self.engine.hide_parameters:
                self._log_info(
                    "[%s] %r",
                    stats,
                    sql_util._repr_params(
                        parameters, batches=10, ismulti=context.executemany
                    ),
                )
            else:
                self._log_info(
                    "[%s] [SQL parameters hidden due to hide_parameters=True]"
                    % (stats,)
                )

        evt_handled = False
        try:
            if context.executemany:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_executemany:
                        if fn(cursor, statement, parameters, context):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_executemany(
                        cursor, statement, parameters, context
                    )
            elif not parameters and context.no_parameters:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_execute_no_params:
                        if fn(cursor, statement, context):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_execute_no_params(
                        cursor, statement, context
                    )
            else:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_execute:
                        if fn(cursor, statement, parameters, context):
                            evt_handled = True
                            break
                if not evt_handled:
>                   self.dialect.do_execute(
                        cursor, statement, parameters, context
                    )

/Users/brapaport/Library/Caches/pypoetry/virtualenvs/app-wQsFoDFt-py3.9/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1799:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x1044cebe0>
cursor = <cursor object at 0x1043069a0; closed: -1>
statement = '\nCREATE TABLE identity (\n\ttax_id_type taxidtype, \n\tfunding_source fundingsource[], \n\tgiven_name VARCHAR NOT NU...T, \n\tid UUID NOT NULL, \n\tuser_id UUID, \n\tPRIMARY KEY (id), \n\tFOREIGN KEY(user_id) REFERENCES users (id)\n)\n\n'
parameters = {}, context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x10452e2e0>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       psycopg2.errors.UndefinedObject: type "taxidtype" does not exist
E       LINE 3:  tax_id_type taxidtype,
E                            ^

/Users/brapaport/Library/Caches/pypoetry/virtualenvs/app-wQsFoDFt-py3.9/lib/python3.9/site-packages/sqlalchemy/engine/default.py:717: UndefinedObject

The above exception was the direct cause of the following exception:

postgresql = <connection object at 0x104333eb0; dsn: 'user=postgres password=xxx dbname=tests host=127.0.0.1 port=17942 options=''', closed: 0>

    @pytest.fixture(name="session")
    def session_fixture(postgresql: connection):
        engine = create_engine(
            "postgresql+psycopg2://",
            connect_args=postgresql.get_dsn_parameters(),
            pool_pre_ping=True,
        )

>       SQLModel.metadata.create_all(engine)

tests/utilities/test_db.py:14:

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.4

Python Version

3.9.7

Additional Context

No response

[yasamoka]

This is related to SQLAlchemy, not to SQLModel.

[invokermain]

I don't think this is a SQL alchemy issue at all, I think it's related to: #164

[olafdeleeuw]

It seems this issue is indeed related to #164. I'm having the same issue when spinning up a Postgres db with testcontainers. For example:

import enum

from sqlalchemy.sql.schema import Column
from sqlmodel import SQLModel, create_engine, Field
from sqlmodel import Enum
from testcontainers.postgres import PostgresContainer
from typing import Optional


class SecretName(enum.Enum):
    S1 = "very_secret_hero1"
    S2 = "very_secret_hero1"


class HeroBase(SQLModel):
    name: str
    secret_name: SecretName = Field(sa_column=Column(Enum(SecretName)))
    age: Optional[int] = None


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


pg_container = PostgresContainer('postgres:latest')
pg_container.start()
engine = create_engine(pg_container.get_connection_url())
SQLModel.metadata.create_all(engine)

results in:

psycopg2.errors.UndefinedObject: type "secretname" does not exist
LINE 3:  secret_name secretname, 

Is there a solution or workaround?

[olafdeleeuw]

Additional info: this only happens when the tables in a database don't exist . So like the example above when you spin up a new postgres db for unittesting.

Let's assume I have two local databases db_heroes and db_heroes2. The database db_heroes has a table as described in the SQLModel above and the database db_heroes2 is empty. The code below works fine for db_heroes and fails for db_heroes2.

However, when I change the type of secret_name into str instead of the Enum, then the code runs smoothly for both databases.

import enum

from sqlmodel import SQLModel, create_engine, Field, Enum
from typing import Optional


class SecretName(enum.Enum):
    S1 = "very_secret_hero1"
    S2 = "very_secret_hero1"


class HeroBase(SQLModel):
    name: str
    secret_name: SecretName = Field(sa_column=Column(Enum(SecretName)))
    age: Optional[int] = None


class Hero(HeroBase, table=True):
    __tablename__ = "heroes"
    id: Optional[int] = Field(default=None, primary_key=True)


engine = create_engine("postgresql://<username>@/<db_name>")
SQLModel.metadata.create_all(engine)