Issue 1:
Describe the bug

I am trying to migrate some ORM classes with mixins and multiple inheritances across the classes to a PostgreSQL DB using the alembic migration tool and I keep getting an error that a "Table" is already defined for a Metadata instance.

To Reproduce

import datetime as dt
from dateutil import tz
import logging
from enum import Enum, IntFlag
from typing import Sequence
from typing import Union, Optional, List, Any, Mapping, Dict

from pydantic import EmailStr, BaseModel as PydanticModel
from sqlalchemy import select, or_, Column, String, Integer, DateTime, Boolean, ForeignKey, MetaData
from sqlalchemy.orm import relationship, registry

from db.db import Base, get_session
from db.adapter import parse_to_schema
from shoppersbag_loyalty import settings
from utils import current_timezone, generate_token, generate_slug

mapper_registry = registry()
logger = logging.getLogger(__name__)

public_models = settings.PUBLIC_MODELS

class Gender(Enum):
MALE = ('M', 'Male')
FEMALE = ('F', 'Female')
UNSAID = ('U', 'Unsaid')

class Actions(IntFlag):
CAN_CREATE = 1
CAN_UPDATE = 2
CAN_LIST = 3
CAN_DESTROY = 5
CAN_RETRIEVE = 7
CAN_READ = 11
CAN_WRITE = 13

def __str__(self):
if self in Actions.__members__:
return str(self.name).lower()
else:
return super(Actions, self).__str__()

class PermissionAction(IntFlag):
DENY_ALL = 0
ALLOW_ANY = 1
ALLOW_ALL = 9
ALLOW_SELECTED_USER = 3
ALLOW_OWNER = 5

def __str__(self):
return self.value

def __eq__(self, other):
if isinstance(other, int):
return self.value == other
else:
return super(PermissionAction, self).__eq__(other)

def as_int(self) -> int:
return self._value_

class UserBase:
__abstract__ = True
metadata = MetaData()

@mapper_registry.mapped
class User(UserBase):
__tablename__ = 'user'

id: int = Column(Integer, primary_key=True)
token: str = Column(String(256), index=True)
slug: str = Column(String, index=True)
email: EmailStr = Column(String, index=True)
mobile: str = Column(String(15))
password: Union[str, bytes] = Column(String(256))
gender: Union[Gender, str] = Column(String(1))
first_name: str = Column(String(30), nullable=False)
last_name: str = Column(String(150), nullable=False)
middle_name: Optional[str] = Column(String(50))
is_business_admin: bool = Column(Boolean, default=False)
username: str = Column(String(150), index=True)
is_staff: bool = Column(Boolean, default=False)
is_admin: bool = Column(Boolean, default=False)
is_active: bool = Column(Boolean, default=True)
date_joined: dt.datetime = Column(DateTime, default=current_timezone)
# relationship fields
groups: List['Group'] = relationship('Group', back_populates='users', secondary='UserGroup')
permissions: List['Permission'] = relationship('Permission', back_populates='users', secondary='UserPermission')
preferences: List['UserPreferences'] = relationship('UserPreferences', back_populates='user')
__table_args__ = {'extend_existing': True}

# @classmethod
# def from_dict(cls, data: Dict):
# try:
# instance = cls(**data)
# return instance
# except Exception:
# raise Exception('Unable to create schema using provided data')

# @classmethod
# def from_schema(cls, data: Any):
# if isinstance(data, PydanticModel):
# return cls.from_dict(data.schema())

# @classmethod
# def from_queryset(cls, queryset: Any) -> Union[Sequence[PydanticModel], None]:
# if isinstance(queryset, Sequence):
# return [parse_to_schema(cls, x) for x in queryset if isinstance(x, (PydanticModel, Dict, Mapping))]

class Permission(Base):
"""
Model a permission system that defines access control to a model (via model field) or
an instance of a model (via model_id) field.
A given action that can be taken by a user or group of users are specified by Actions enum.
"""
__tablename__ = 'permission'

id: int = Column(Integer, primary_key=True)
token: str = Column(String(256), index=True)
slug: str = Column(String, index=True)
name: str = Column(String, nullable=False)
action: Actions = Column(Integer, nullable=False)
model_id: Optional[int] = Column(Integer)
model: Optional[str] = Column(String(256)) # model name in form db.<package>.ClassName
users: List[User] = relationship(User, back_populates='permissions', secondary='UserPermission')
groups: List['Group'] = relationship('Group', back_populates='', secondary='GroupPermission')

__table_args__ = {'extend_existing': True}

class Group(Base):
__tablename__ = 'group'

id: int = Column(Integer, primary_key=True)
token: str = Column(String(256), index=True)
slug: str = Column(String, index=True)
name: str = Column(String, nullable=False)
permissions: List[Permission] = relationship(Permission, back_populates='groups', secondary='GroupPermission')
users: List[User] = relationship(User, back_populates='groups', secondary='UserGroup')

class GroupPermission(Base):
__tablename__ = 'group_permission'

group_id: Optional[int] = Column(ForeignKey("group.id"), primary_key=True)
permission_id: Optional[int] = Column(ForeignKey("permission.id"), primary_key=True)

class UserPermission(Base):
__tablename__ = 'user_permission'

user_id: Optional[int] = Column(ForeignKey("user.id"), primary_key=True)
permission_id: Optional[int] = Column(ForeignKey("permission.id"), primary_key=True)

class UserGroup(Base):
__tablename__ = 'user_group'

user_id: Optional[int] = Column(ForeignKey("user.id"), primary_key=True)
group_id: Optional[int] = Column(ForeignKey("group.id"), primary_key=True)

class ModelPermissionMixin:
__abstract__ = True
__READ__ = (Actions.CAN_CREATE, Actions.CAN_UPDATE, Actions.CAN_DESTROY, Actions.CAN_WRITE,
Actions.CAN_READ, Actions.CAN_RETRIEVE)
__WRITE__ = (Actions.CAN_WRITE, Actions.CAN_DESTROY, Actions.CAN_UPDATE)
__DESTROY__ = (Actions.CAN_DESTROY,)

id = Column(Integer, primary_key=True)
token = Column(String(256), index=True, default=generate_token)
slug = Column(String, default=generate_slug)

metadata = MetaData()

@property
def pk(self):
return self.id

def _process_object_permission(self, user, checks):
session_local = get_session()
with session_local.begin() as session:
ps = session.execute(select(Permission)).all()
acts = any([bool(x) for x in ps if user in x.users and x.model_id == self.pk and x.action in checks])
return acts or any([x for x in public_models if x == self.__name__])

@classmethod
def _process_class_permission(cls, user, checks):
session_local = get_session()
with session_local.begin() as session:
ps = session.execute(select(Permission)).all()
acts = any([bool(x) for x in ps if user in x.users and x.model == cls.__name__ and x.action in checks])
return acts or any([x for x in public_models if x == cls.__name__])

def has_object_read_permission(self, user):
# Fetch model owner from Permission table and validate if user matched parameter
return self._process_object_permission(user, self.__READ__)

@classmethod
def has_read_permission(cls, user):
return cls._process_class_permission(user, cls.__READ__)

def has_object_write_permission(self, user):
return self._process_object_permission(user, self.__WRITE__)

@classmethod
def has_write_permission(cls, user):
return cls._process_class_permission(user, cls.__WRITE__)

class UserPreferences(Base):
__tablename__ = 'user_preferences'

id: int = Column(Integer, primary_key=True)
token: str = Column(String(256), index=True)
slug: str = Column(String, index=True)
key: str = Column(String)
value: str = Column(String)
user: Optional[User] = relationship(User, back_populates='preferences')
user_id: Optional[int] = Column(Integer, ForeignKey('user.id'))

def get_user_by_identity(identity: Any) -> Any:
session_local = get_session()
with session_local.begin() as session:
return session.execute(select(User).where(
or_(User.slug == identity, User.token == identity, User.id == identity))).first()

class UserAccess(Base):
__tablename__ = 'user_access'

id = Column(Integer, primary_key=True)
token = Column(String(256), index=True)
slug = Column(String)
user_id = Column(Integer, ForeignKey('user.id'), default=None)
user = relationship(User)
start = Column(DateTime)
end = Column(DateTime)
expired = Column(Boolean, default=False)

def __init__(self, start: dt.datetime, end: dt.datetime = None):
# if the current time exceeds duration of token set end time to current

local_timezone = tz.tzlocal()
self.start = start.astimezone(tz=local_timezone)
if end:
self.end = end.astimezone(tz=local_timezone)
else:
self.end = self.start + dt.timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)

now = dt.datetime.now(tz=local_timezone)
self.expired = (self.end - self.start < dt.timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)) or \
(now - self.start < dt.timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
super(UserAccess, self).__init__()

@property
def current_user(self):
return self.user if self.end - self.start < dt.timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) else None

class UserVerification(Base):
__tablename__ = 'user_verification'

id = Column(Integer, primary_key=True)
token = Column(String(256), index=True)
slug = Column(String)
verified_on = Column(DateTime)
verified_by = Column(String)
verification_document = Column(String)
is_approved = Column(Boolean, default=False)
user_id = Column(Integer, ForeignKey('user.id'), default=None)
user = relationship(User)

Error

# Copy complete stack trace and error message here, including SQL log output if applicable.
``` File "C:\Users\User\AppData\Local\Programs\Python\Python310\lib\runpy.py", line 196, in _run_module_as_main
return _run_code(code, main_globals, None,
File "C:\Users\User\AppData\Local\Programs\Python\Python310\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\Scripts\alembic.exe\__main__.py", line 7, in <module>
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\alembic\config.py", line 588, in main
CommandLine(prog=prog).main(argv=argv)
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\alembic\config.py", line 582, in main
self.run_cmd(cfg, options)
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\alembic\config.py", line 559, in run_cmd
fn(
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\alembic\command.py", line 227, in revision
script_directory.run_env()
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\alembic\script\base.py", line 563, in run_env
util.load_python_file(self.dir, "env.py")
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\alembic\util\pyfiles.py", line 92, in load_python_file
module = load_module_py(module_id, path)
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\alembic\util\pyfiles.py", line 108, in load_module_py
spec.loader.exec_module(module) # type: ignore
File "<frozen importlib._bootstrap_external>", line 883, in exec_module
table_cls(
File "<string>", line 2, in __new__
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\sqlalchemy\util\deprecations.py", line 309, in warned return fn(*args, **kwargs)
File "C:\Users\User\Desktop\Coding\Shoppers_Bag\SHOPERS_BAG_FASTAPI\venv\lib\site-packages\sqlalchemy\sql\schema.py", line 593, in __new__
raise exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Table 'permission' is already defined for this MetaData instance. Specify 'extend_existing=True' to redefine options and columns on an existing Table object.

### Versions

- OS:
- Python:
- SQLAlchemy:
- Database:
- DBAPI (eg: psycopg, cx_oracle, mysqlclient):

### Additional context

_No response_

Issue 2:
Migrated issue, originally created by Michael Bayer (@zzzeek)

have the polymorphic_on check during loading continue to check on the type returned so that polymorphic loads can cascade. Other complexities here regard getting the multiple discriminators assigned during init, as well as considering if polymorphic_map can be local to each sub-hierarchy.

"workaround" example below.

This is only tentatively on the 0.8 milestone. can move to 0.9.

"""
mixed single and joined table inheritance.
"""

from sqlalchemy import *
from sqlalchemy import types
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy import event

Base = declarative_base()

class Product(Base):
__tablename__ = 'products'
id = Column(types.Integer, primary_key=True)
discriminator = Column('product_type', types.String(50), nullable=False)

_discriminator = "discriminator"

def price_history(self):
return [PhysicalProduct(Product):
p_discr = Column(types.String(50))

_discriminator = "p_discr"

@declared_attr
def __mapper_args__(cls):
return {'polymorphic_identity': 'physical_product'}

def inventory(self):
return "computed inventory"

class NonPhysicalProduct(Product):
np_discr = Column(types.String(50))

_discriminator = "np_discr"

@declared_attr
def __mapper_args__(cls):
return {'polymorphic_identity': 'nonphysical_product'}

def somefunc(self):
return "someval"

# set polymorphic on as a coalesce of those three
# columns. It's after the fact beacuse p_discr and np_discr
# are defined after Product, but if you move them up then
# this can be inline inside of Product.__mapper_args__.
# this would improve loads too as it appears the p_discr/np_discr columns
# aren't loaded directly when you query for Product
for mp in Product.__mapper__.self_and_descendants:
mp._set_polymorphic_on(
func.coalesce(
Product.__table__.c.p_discr,
Product.__table__.c.np_discr,
Product.__table__.c.product_type
))

# build our own system of assigning polymorphic identities
# to instances; use the 'init' event.
# Add a "print" for the "identity" dict to see what it's doing.
@event.listens_for(Product, "init", propagate=True)
def init(target, args, kwargs):
identity = {}
for cls, supercls in zip(type(target).__mro__, type(target).__mro__[1:](]

class)):
if not hasattr(supercls, '_discriminator'):
break
discriminator_attr = supercls._discriminator
poly_identity = cls.__mapper__.polymorphic_identity
identity.setdefault(discriminator_attr, poly_identity)
for key in identity:
setattr(target, key, identity[key](key))

class Newspaper(PhysicalProduct):
__tablename__ = 'newspapers'
__mapper_args__ = {'polymorphic_identity': 'newspaper'}

id = Column(types.Integer,
ForeignKey('products.id'),
primary_key=True
)
title = Column(types.String(50))

def __init__(self, title):
self.title = title

class NewspaperDelivery(NonPhysicalProduct):
__tablename__ = 'deliveries'
__mapper_args__ = {'polymorphic_identity': 'delivery'}

id = Column(types.Integer,
ForeignKey('products.id'),
primary_key=True
)
destination = Column(types.String(50))

def __init__(self, destination):
self.destination = destination

# note here how the polymorphic map works out:
print Product.__mapper__.polymorphic_map
# {'newspaper': <Mapper at 0x1014d8890; Newspaper>,
# 'delivery': <Mapper at 0x1014dec90; NewspaperDelivery>,
# 'nonphysical_product': <Mapper at 0x1014d2350; NonPhysicalProduct>,
# 'physical_product': <Mapper at 0x1014d00d0; PhysicalProduct>}

e = create_engine('sqlite:///:memory:', echo='debug')
Base.metadata.drop_all(e)
Base.metadata.create_all(e)

session = Session(e, autoflush=True, autocommit=False)

session.add_all([ Newspaper(title="Financial Times"),
NewspaperDelivery(destination="__somewhere__"),
PhysicalProduct(),
NonPhysicalProduct()
](
))

session.commit()

# the important part - that a row only known as Product can
# interpret as a specific subclass
assert [ type(c) for c in session.query(Product).order_by(Product.id)
](
) == [NewspaperDelivery, PhysicalProduct, NonPhysicalProduct](Newspaper,)

# test sub-table load. The load for "title" apparently emits a JOIN still because
# in order to refresh the subclass of "Product" it also wants to get
# at p_discr.
np = session.query(Product).filter_by(id=1).first()
assert np.title == "Financial Times"

session.close()

# in this version, it emits two separate, single table SELECT statements,
# since the first query loads the full set of columns for PhysicalProduct.
np = session.query(PhysicalProduct).filter_by(id=1).first()
assert np.title == "Financial Times"