Hi there -
I normally wouldn't do things this way, however in seeking to provide
an alternative architecture, the specific example you've given does
work when used with the correct constructs, so I can provide both
architectures. So you would need to illustrate specific scenarios
that aren't working in terms of a fully runnable example, e.g. the one
below. The awkardness of doing it this way is that ORM declarative is
mapping each subclass individually, using single table inheritance
without a polymorphic_identity. So when you for example query for
Location_lvlB and then later query for Location_lvlA, you can get the
*same* row but in two different object instances. This creates a
conflict against the row which while it might be something you can
work with, the ORM is not aware that it's working this way:
(Pdb) l1 = session.query(Location_lvlA).first()
(Pdb) l2 = session.query(Location_lvlB).first()
(Pdb) l1
<__main__.Location_lvlA object at 0x7f274c47a940>
(Pdb) l2
<__main__.Location_lvlB object at 0x7f274c47abe0>
The second example attached illustrates a safer way to go which is
that you only map one class per hiearchy, the bottommost one, and then
express your various levels of class functionality in terms of mixins.
Hope this helps.
# example 1, the requested architecture
import datetime as dt
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import DateTime
from sqlalchemy import ForeignKey
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import Sequence
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm.session import object_session
metadata = MetaData()
source_table = Table(
"source",
metadata,
Column(
"id",
Integer,
Sequence("src_id_seq", metadata=metadata),
primary_key=True,
),
Column("name", String(length=32), unique=True, nullable=False, index=True),
Column(
"created",
DateTime(timezone=False),
default=dt.datetime.utcnow,
nullable=False,
),
)
location_table = Table(
"auth_user_detail",
metadata,
Column(
"id",
Integer,
Sequence("loc_id_seq", metadata=metadata),
primary_key=True,
),
Column("name", String(length=32), unique=True, nullable=False, index=True),
Column(
"created",
DateTime(timezone=False),
default=dt.datetime.utcnow,
nullable=False,
),
Column("source_id", ForeignKey("source.id")),
Column("firstname", String),
)
# base definition of the Source-class for Mapper
Base = declarative_base(metadata=metadata)
class Source_orm(Base):
__table__ = source_table
_loc = relationship("Location_orm", uselist=False)
loc_name = association_proxy("_loc", "firstname")
def __init__(self, name):
self.name = name
# base definition of the Location-class for Mapping
class Location_orm(Base):
__table__ = location_table
def __init__(self, name):
self.name = name
# -------------------
# Higher functions - lvlA : possibly packed into a different module
class Source_lvlA(Source_orm):
@classmethod
def get_by_name(cls, session, name):
return session.query(cls).filter(cls.name == name).one()
def move_to_loc_by_name(self, loc_name):
session = object_session(self)
loc = (
session.query(Location_orm)
.filter(Location_orm.name == loc_name)
.one()
)
self._loc = loc
session.commit()
class Location_lvlA(Location_orm):
@classmethod
def get_by_name(cls, session, name):
return session.query(cls).filter(cls.name == name).one()
def move_src_here(self, src):
session = object_session(self)
src.loc_id = self.id
session.merge(src)
session.commit()
# -------------------
# Even Higher functions - lvlB : possibly packed into a different module
class Source_lvlB(Source_lvlA):
def assemble_info(self):
return f"<Source> {self.name} at <Location> {self.loc_name}"
class Location_lvlB(Location_lvlA):
def assemble_info(self):
return f"<Location> {self.name}"
if __name__ == "__main__":
engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)
session = Session(engine)
# create low level objects
s = Source_lvlA("MySource")
session.add(s)
session.flush()
l = Location_lvlB("MyLocation")
session.add(l)
session.flush()
# operate on the db use a higher level function
s = Source_lvlA.get_by_name(session, "MySource")
s.move_to_loc_by_name(
Location_lvlB.get_by_name(session, "MyLocation").name
)
# use highest level functionality
s = Source_lvlB.get_by_name(session, "MySource").assemble_info()
# example 2 - the safer architecture
import datetime as dt
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import DateTime
from sqlalchemy import ForeignKey
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import Sequence
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm.session import object_session
metadata = MetaData()
source_table = Table(
"source",
metadata,
Column(
"id",
Integer,
Sequence("src_id_seq", metadata=metadata),
primary_key=True,
),
Column("name", String(length=32), unique=True, nullable=False, index=True),
Column(
"created",
DateTime(timezone=False),
default=dt.datetime.utcnow,
nullable=False,
),
)
location_table = Table(
"auth_user_detail",
metadata,
Column(
"id",
Integer,
Sequence("loc_id_seq", metadata=metadata),
primary_key=True,
),
Column("name", String(length=32), unique=True, nullable=False, index=True),
Column(
"created",
DateTime(timezone=False),
default=dt.datetime.utcnow,
nullable=False,
),
Column("source_id", ForeignKey("source.id")),
Column("firstname", String),
)
# base definition of the Source-class for Mapper
Base = declarative_base(metadata=metadata)
class Source_orm(Base):
__abstract__ = True
@declared_attr
def __table__(cls):
return source_table
@declared_attr
def _loc(cls):
return relationship("Location_mapped", uselist=False)
loc_name = association_proxy("_loc", "firstname")
def __init__(self, name):
self.name = name
# base definition of the Location-class for Mapping
class Location_orm(Base):
__abstract__ = True
@declared_attr
def __table__(cls):
return location_table
def __init__(self, name):
self.name = name
# -------------------
# Higher functions - lvlA : possibly packed into a different module
class Source_abstract:
pass
class Location_abstract:
pass
class Source_lvlA(Source_abstract):
@classmethod
def get_by_name(cls, session, name):
return session.query(cls).filter(cls.name == name).one()
def move_to_loc_by_name(self, loc_name):
Location_mapped = inspect(self).mapper.attrs._loc.mapper.class_
session = object_session(self)
loc = (
session.query(Location_mapped)
.filter(Location_mapped.name == loc_name)
.one()
)
self._loc = loc
session.commit()
class Location_lvlA(Location_abstract):
@classmethod
def get_by_name(cls, session, name):
return session.query(cls).filter(cls.name == name).one()
def move_src_here(self, src):
session = object_session(self)
src.loc_id = self.id
session.merge(src)
session.commit()
# -------------------
# Even Higher functions - lvlB : possibly packed into a different module
class Source_lvlB(Source_lvlA):
def assemble_info(self):
return f"<Source> {self.name} at <Location> {self.loc_name}"
class Location_lvlB(Location_lvlA):
def assemble_info(self):
return f"<Location> {self.name}"
class Source_mapped(Source_lvlB, Source_orm):
pass
class Location_mapped(Location_lvlB, Location_orm):
pass
if __name__ == "__main__":
engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)
session = Session(engine)
# create low level objects
s = Source_mapped("MySource")
session.add(s)
session.flush()
l = Location_mapped("MyLocation")
session.add(l)
session.flush()
# operate on the db use a higher level function
s = Source_mapped.get_by_name(session, "MySource")
s.move_to_loc_by_name(
Location_mapped.get_by_name(session, "MyLocation").name
)
# use highest level functionality
s = Source_mapped.get_by_name(session, "MySource").assemble_info()
On Tue, May 14, 2019 at 11:25 AM Marcel Zoll
<[email protected]> wrote:
>
> Hi, I am struggling since some days to extend and implement common class
> inheritance for the SQLAlchemy ORM objects that I define: Here is the to be
> achieved task:
> 1. In a first step, I want to use sqlalchemy to model my database and the
> ORM, aka define tables, define the ORMs and their relations and some extras
> (association mapper) in Order to make my live easier
> 2. In a second step I want to take the ORMs and extend the classes with some
> higher level functionality, modeling common operations which will be executed
> in the database.
> 3. In a third step, I'd like to take the extended ORMs from step 2 and extend
> them again with further convenience functions, which are, however, separate
> from the db-model logic.
>
> To logically isolate and structure the whole thing, I want to wrap each of
> these steps into a dedicated module/package. This I would have solved by
> class Inheritance, which seams to be the natural approach.
>
> However, for all methods which I tried, I cannot figure out how this done
> correctly even in mock scripts. I will post below some example code which
> should demonstrate what I want to achieve.
>
> I tried multiple approaches:
> - with classical mapping (wont work, because i need to explicitly map every
> derived class again, copying all mapper attributes, I would have to double
> paste a lot of code, which is exactly not the point of inheritance)
> - with declarative (there is some deep magic to these
> @declarative_attributes, which I do not understand when to use and when not,
> respective at which point they are resolved)
> - with hybrid approach, my currently preferred one, as I have at least the
> table definitions done, before messing with the ORM
> I could for all of them not find a single solution which would fulfill all my
> needs or which do not throw errors:
> Errors include:
> - The derived class-names cannot be resolved for the relation to work
> correctly
> - The attributes (relation, association_proxy) do not propagate up to the
> derived classes (I have the feeling they bind to the first concrete mapped
> class they are in and inheritance onward is impossible)
>
> If anybody could give me some pointers or make the below code example work by
> some magic alchemistic conjurement I would be very grateful.
>
> Marcel
>
> ==================Example Code ===================
> import datetime as dt
>
> from sqlalchemy.ext.declarative import declarative_base
> from sqlalchemy import MetaData, Table, Column, Sequence, ForeignKey
> from sqlalchemy import Integer, String, DateTime
>
> from sqlalchemy.orm import relationship
> from sqlalchemy.ext.associationproxy import association_proxy
>
> from sqlalchemy.orm import Session
> from sqlalchemy import create_engine
> from sqlalchemy.orm.session import object_session
>
> metadata = MetaData(schema='sandbox')
>
> source_table = Table('source', metadata,
> Column('id', Integer, Sequence("src_id_seq", metadata=metadata),
> primary_key=True),
> Column('name', String(length=32), unique=True, nullable=False,
> index=True),
> Column('created', DateTime(timezone=False), default=dt.datetime.utcnow,
> nullable=False),
> Column('loc_id', ForeignKey(user_table.c.id), nullable=True))
>
> location_table = Table('auth_user_detail', metadata,
> Column('id', Integer, Sequence("loc_id_seq", metadata=metadata),
> primary_key=True),
> Column('name', String(length=32), unique=True, nullable=False,
> index=True),
> Column('created', DateTime(timezone=False), default=dt.datetime.utcnow,
> nullable=False))
> )
>
> # base definition of the Source-class for Mapper
> Base = declarative_base(metadata)
>
> class Source_orm(Base):
> __table__ = source_table
> _loc = relationship('Location', uselist=False)
> loc_name = association_proxy('_loc', 'firstname')
>
> def __init__(self, name):
> self.name = name
>
> # base definition of the Location-class for Mapping
> class Location_orm(Base):
> __table__ = location_table
>
> def __init__(self, name):
> self.name = name
>
>
> #-------------------
> # Higher functions - lvlA : possibly packed into a different module
>
> class Source_lvlA(Source_orm):
> @classmethod
> def get_by_name(cls, session, name):
> return session.query(cls).filter(cls.name == name).one()
>
> def move_to_loc_by_name(self, loc_name):
> session = object_session(self)
> loc = session.query(Location).filter(Location.name == loc_name).one()
> self._loc = loc
> session.commit()
>
> class Location_lvlA(Location_orm):
> @classmethod
> def get_by_name(cls, session, name):
> return session.query(cls).filter(cls.name == name).one()
>
> def move_src_here(self, src):
> session = object_session(self)
> src.loc_id = self.id
> session.merge(src)
> session.commit()
>
> #-------------------
> # Even Higher functions - lvlB : possibly packed into a different module
>
> class Source_lvlB(Source_lvlA):
> def assemble_info(self):
> return f"<Source> {self.name} at <Location> {self.loc_name}"
>
> class Location_lvlB(Location_lvlA):
> def assemble_info(self):
> return f"<Location> {self.name}"
>
>
> if __name__ == '__main__':
> engine = create_engine("sqlite://", echo=True)
> engine.execute("""DROP SCHEMA IF EXISTS {schema} CASCADE;
> """.format(schema=metadata.schema))
> engine.execute("""CREATE SCHEMA
> {schema};""".format(schema=metadata.schema))
>
> Base.metadata.create_all(engine)
> session = Session(engine)
>
> # create low level objects
> s = Source_ormA('MySource')
> s = session.add(a)
> session.flush()
> l = Location_lvlA('MyLocation')
> l = session.add(l)
> session.flush()
>
> # operate on the db use a higher level function
> s = Source_lvlA.get_by_name(session, 'MySource')
> s.move_to_loc(Location_lvlB.get_by_name('MyLocation'))
>
> # use highest level functionality
> s = Source_lvlB.get_by_name(session, 'MySource').assemble_info()
>
> --
> SQLAlchemy -
> The Python SQL Toolkit and Object Relational Mapper
>
> http://www.sqlalchemy.org/
>
> To post example code, please provide an MCVE: Minimal, Complete, and
> Verifiable Example. See http://stackoverflow.com/help/mcve for a full
> description.
> ---
> You received this message because you are subscribed to the Google Groups
> "sqlalchemy" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to [email protected].
> To post to this group, send email to [email protected].
> Visit this group at https://groups.google.com/group/sqlalchemy.
> To view this discussion on the web visit
> https://groups.google.com/d/msgid/sqlalchemy/ed3a4914-2b37-4f2e-b213-3c615b85c367%40googlegroups.com.
> For more options, visit https://groups.google.com/d/optout.
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
To view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/CA%2BRjkXGVzoGtxD7xotYiNY2juvoG3Zxd4timj3o4sZcfX2vxWQ%40mail.gmail.com.
For more options, visit https://groups.google.com/d/optout.
