Skip to content

psycopg2.errors.UndefinedObject: type MY_ENUM does not exist #131

Description

@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

Activity

  1. yasamoka commented on Oct 14, 2021

    @yasamoka

    This is related to SQLAlchemy, not to SQLModel.

  2. invokermain commented on Apr 11, 2022

    @invokermain

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

  3. olafdeleeuw commented on Jun 22, 2022

    @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?

  4. olafdeleeuw commented on Jun 23, 2022

    @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)
    
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    questionFurther information is requested

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions