Re: [sqlalchemy] Can I access session.identity_map? Or is there a better way to do this?
The reason why I used a in-database cache is that the group permissions are not often changed, so when a user logs in, I can know its permissions without querying the other three tables. And I used a in-memory cache because it's quite frequent to check user permissions, so I do not have to json decode the string each time. But your session-based cache solution is really cool! :) Thanks very much for your kind reply! -- You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To view this discussion on the web visit https://groups.google.com/d/msg/sqlalchemy/-/63IIKeHVnuYJ. To post to this group, send email to sqlalchemy@googlegroups.com. To unsubscribe from this group, send email to sqlalchemy+unsubscr...@googlegroups.com. For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en.
[sqlalchemy] Set-based association proxy through AppenderQuery?
Hi,
Is it possible to have an association_proxy (in the association object
pattern) that emulates a set-based collection if it goes through a
lazy='dynamic' relationship? I can't for the life of me find a way to
make this work (setting collection_class on the dynamic relationship
doesn't seem to do anything).
Here's some example code of what I'm trying to do, extracted from the
actual project:
class ProofOfWork(object):
blocks = association_proxy('Intermediatory_nodes', 'block')
proof_of_work = Table('proof_of_work', db.metadata)
mapper(ProofOfWork, proof_of_work, properties={
'Intermediatory_nodes': relationship(lambda: Intermediatory,
lazy = 'dynamic'),
})
class Block(object):
proof_of_works = association_proxy('Intermediatory_nodes',
'proof_of_work')
block = Table('block', db.metadata)
mapper(Block, block, properties={
'Intermediatory_nodes': relationship(lambda: Intermediatory,
lazy = 'dynamic'),
})
class Intermediatory(object):
pass
intermediatory = Table('intermediatory', db.metadata,
Column('proof_of_work_id', Integer,
ForeignKey('proof_of_work.id'),
nullable = False),
Column('block_id', Integer,
ForeignKey('block.id')),
)
mapper(Intermediatory, intermediatory, properties={
'proof_of_work': relationship(lambda: ProofOfWork,
back_populates = 'Intermediatory_nodes',
remote_side= lambda: proof_of_work.c.id),
'block': relationship(lambda: Block,
back_populates = 'Intermediatory_nodes',
remote_side= lambda: block.c.id),
})
How can I make ProofOfWork.blocks and Block.proof_of_works return an
_AssociationSet instead of _AssociationList?
Cheers,
Mark
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
Re: [sqlalchemy] HOW TO HANDLE SPECIAL CHARACTERS WITH ENGINE.EXECUTE
Thanks Michael,
I will see if this works perfectly with postgresql.
I had tryed func before but did not get any success.
May be this time it will work.
Happy hacking.
Krishnakant.
On 11/02/12 21:10, Michael Bayer wrote:
On Feb 11, 2012, at 8:02 AM, Krishnakant Mane wrote:
Hello all,
I have an interesting problem for which I am sure some simple solution must be
existing.
I have made a Python function which I will paist below.
Basically what the function does is that it takes 3 parameters, namely the name
of a stored procedure, engine instance and a set of parameters.
This function is used to make calls to postgresql based stored procedures in a
modular way.
The function is kept central and all my modules just pass the necessary
parameters and leave it to the function to do the rest.
I got it working perfectly, except that I don't know how to handle special
characters when constructing the query that makes a call to a stored procedure.
So if I have an insert query which has a value with a single quote ('), it
crashes.
This is because the function is not using bound parameters. Dealing with
individual datatypes and how they are formatted to the database is something
you should let the DBAPI handle.
SQLAlchemy includes the capability to call functions built in via the "func"
parameter. Your execproc could be written as:
from sqlalchemy import create_engine, func
engine = create_engine('mysql://root@localhost/test', echo=True)
def execproc(procname, engine, queryParams=[]):
function = getattr(func, procname)
function_with_params = function(*queryParams)
return
engine.execute(function_with_params.execution_options(autocommit=True))
print execproc("concat", engine, ["dog", " ", "cat"]).scalar()
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
Re: [sqlalchemy] Delete failing with StaleDataError
On Saturday 11 Feb 2012 10:32:12 AM Michael Bayer wrote: > I ran this and indeed InnoDB appears to be broken here. This is a MySQL > bug. It would appear it is running the cascade between the two "A" rows > and only considering the lead object to be the one deleted.Can't > exactly find it at bugs.mysql.com either, so you'd do everyone a favor if > you could create a ticket over there. I've raised this on lists.mysql. > > SQLAlchemy can try to work around this but for now you can just flip > supports_sane_rowcount off for the whole dialect: Thanks, that would be a lot better than writing a new migrate script. -- Fayaz Yusuf Khan Cloud developer and architect Dexetra SS, Bangalore, India fayaz.yusuf.khan_AT_gmail_DOT_com fayaz_AT_dexetra_DOT_com +91-9746-830-823 signature.asc Description: This is a digitally signed message part.
Re: [sqlalchemy] Can I access session.identity_map? Or is there a better way to do this?
On Feb 11, 2012, at 7:04 AM, Kai Jia wrote:
> Hi,
> The situation is that I have tow tables (User and UserGrp) and two
> association tables (one for many-to-many between User and UserGrp, one for
> many-to-many between UserGrp and user group permissions). Each user can
> belong to multiple user groups, and each user group can have multiple
> permissions. The permissions that a user has is the union of the permissions
> of the groups that it belongs to.
>
> I cached the permission of each user in the User table (see the
> User._perms_cache attribute). So it is necessary to invalidate the cache when
> the relationship between users and user groups gets changed. To invalidate
> the cache, I do an UPDATE on all the affected users, and I also have to
> expire the _perms_cache attribute of all persistent User instances. However,
> I searched the docs and couldn't find an appropriate API (Session.expire_all
> does not work, which is shown later; Session.expire requires an instance).
>
> I looked into SQLAlchemy's source, and finally found that I can iterate over
> all the persistent objects via Session.identity_map. But this is not
> documented,
this blew me away that it didn't make it into the docs, fixed:
http://docs.sqlalchemy.org/en/latest/orm/session.html#session-attributes
http://docs.sqlalchemy.org/en/latest/orm/session.html#sqlalchemy.orm.session.Session.identity_map
> and do I really have to do this?
there's a lot of variables to what you're doing, caching is a pretty open ended
thing.Interesting here that you're caching in two different ways at the
same time, both as a textual column and an in-memory map. I'd probably convert
_perms_cache to use a Json column and just skip the extra in-memory dictionary,
I'm not sure what you're gaining by caching twice like that.
But, for the general issue of a session-scoped, in-memory cache, this is
common.You might want to consider that the problem you're trying to solve
is a per-Session cache of permissions.But when you deal with each User
object, you're storing a cache locally on each User. Why not just stick the
dictionary on the Session itself ?
class User(Base):
#
_perms_cache = Column('permscache', Text())
@property
def perms(self):
sess = object_session(self)
if not hasattr(sess, '_perms_cache'):
sess._perms_cache = {}
if self in sess._perms_cache:
return sess._perms_cache[self]
else:
sess._perms_cache[self] = result = self._get_perms()
return result
# ...
def invalidate_user_perm_cache(session, gid):
try:
del session._perms_cache
except AttributeError:
pass
sub = session.query(MapUserAndUserGrp.uid) \
.filter(MapUserAndUserGrp.gid == gid)
session.query(User).filter(User.id.in_(sub)) \
.update({User._perms_cache: None}, synchronize_session = False)
> The simplified model is attached (sorry...it's still that long)
>
> And I also wonder why the following does not work? It raises an
> IntegrityError.
>
> python test.py
> >>> g0.perms.add(5)
> >>> ses.expire_all()
> >>> ses.commit()
this model took me a long time to get my head around, guess its early since
it's not doing anything weird...guess it's the names.
when you expire_all(), the pending changes on UsrGrp, which include that a new
MapUserGrpAndPerm has been associated with it, is removed. But the actual
MapUserGrpAndPerm remains pending in Session.new. The unit of work flushes
it, the "gid" column fails to get populated since UsrGrp's changes are gone,
then you get a NULL integrity constraint.
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
Re: [sqlalchemy] HOW TO HANDLE SPECIAL CHARACTERS WITH ENGINE.EXECUTE
On Feb 11, 2012, at 8:02 AM, Krishnakant Mane wrote:
> Hello all,
> I have an interesting problem for which I am sure some simple solution must
> be existing.
> I have made a Python function which I will paist below.
> Basically what the function does is that it takes 3 parameters, namely the
> name of a stored procedure, engine instance and a set of parameters.
> This function is used to make calls to postgresql based stored procedures in
> a modular way.
> The function is kept central and all my modules just pass the necessary
> parameters and leave it to the function to do the rest.
> I got it working perfectly, except that I don't know how to handle special
> characters when constructing the query that makes a call to a stored
> procedure. So if I have an insert query which has a value with a single
> quote ('), it crashes.
This is because the function is not using bound parameters. Dealing with
individual datatypes and how they are formatted to the database is something
you should let the DBAPI handle.
SQLAlchemy includes the capability to call functions built in via the "func"
parameter. Your execproc could be written as:
from sqlalchemy import create_engine, func
engine = create_engine('mysql://root@localhost/test', echo=True)
def execproc(procname, engine, queryParams=[]):
function = getattr(func, procname)
function_with_params = function(*queryParams)
return
engine.execute(function_with_params.execution_options(autocommit=True))
print execproc("concat", engine, ["dog", " ", "cat"]).scalar()
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
Re: [sqlalchemy] Delete failing with StaleDataError
-BEGIN PGP SIGNED MESSAGE- Hash: SHA1 On Feb 11, 2012, at 5:22 AM, Fayaz Yusuf Khan wrote: > The attached script fails with this: > Traceback (most recent call last): > File "stale_delete.py", line 33, in >session.flush() > File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", > line 1559, in flush >self._flush(objects) > File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", > line 1630, in _flush >flush_context.execute() > File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/unitofwork.py", > line 331, in execute >rec.execute(self) > File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/unitofwork.py", > line 498, in execute >uow > File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/mapper.py", line > 2507, in _delete_obj >(table.description, len(del_objects), c.rowcount) > sqlalchemy.orm.exc.StaleDataError: DELETE statement on table 'A' expected to > delete 2 row(s); 1 were matched. > > > in MySQL-InnoDB (works in SQLite and Postgres). > Tried versions 0.7.3 and 0.7.5 > Python 2.7 > > I tried manually deleting the rows from the table through MySQL client and > noticed that it's not returning an accurate row count (doesn't > 'supports_sane_rowcount') for a table with an adjacency relationship and an > ondelete='CASCADE'. > > I'm going to work around this by adding another ondelete='CASCADE' for User->A I ran this and indeed InnoDB appears to be broken here. This is a MySQL bug. It would appear it is running the cascade between the two "A" rows and only considering the lead object to be the one deleted.Can't exactly find it at bugs.mysql.com either, so you'd do everyone a favor if you could create a ticket over there. SQLAlchemy can try to work around this but for now you can just flip supports_sane_rowcount off for the whole dialect: engine.dialect.supports_sane_rowcount = engine.dialect.supports_sane_multi_rowcount = False The only solution I can see is another mapper flag that lets you disable the rowcount check for specific classes (see http://www.sqlalchemy.org/trac/ticket/2403) .Unfortunately I don't see a way to make this very obvious to users not aware of the issue. -BEGIN PGP SIGNATURE- Version: GnuPG v1.4.11 (Darwin) Comment: GPGTools - http://gpgtools.org iQEcBAEBAgAGBQJPNon9AAoJEDMCOcHE2v7hVeIH/0HFDYCLUiKH6vuRvEvCapjD XrCwAlpjbu6dCE82YNn00oSvekh0QfFgDgIFafnZzavtE1dn2ve9HvDBviUmmE+h gpMuhVayAZxZfOAYu8512hSb9rx/ZHQv+wt/e4Rl3pnbBGLq1Jg8bufd+S0Ytkwx uIktgEQjfEIRJIfBK0/j+xp8y8lVo/2lFA8or/WdLulgepddgOdzEI6RTkUk45pl Cuq8VVMc49BQaCcgmc0aFmZ9lrhopKii2/31HRca384sk358Otm+9sbomKNdBZbQ QRh+LhqUL53Rgi+69d2vWPfGZMkvE6q4DI89qVdVlzuh1SXI62vwXTRb/rEuEe4= =Q+h5 -END PGP SIGNATURE- -- You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To post to this group, send email to sqlalchemy@googlegroups.com. To unsubscribe from this group, send email to sqlalchemy+unsubscr...@googlegroups.com. For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en.
[sqlalchemy] Can I access session.identity_map? Or is there a better way to do this?
Hi,
The situation is that I have tow tables (User and UserGrp) and two
association tables (one for many-to-many between User and UserGrp, one for
many-to-many between UserGrp and user group permissions). Each user can
belong to multiple user groups, and each user group can have multiple
permissions. The permissions that a user has is the union of the
permissions of the groups that it belongs to.
I cached the permission of each user in the User table (see the *
User._perms_cache* attribute). So it is necessary to invalidate the cache
when the relationship between users and user groups gets changed. To
invalidate the cache, I do an UPDATE on all the affected users, and I also
have to expire the _perms_cache attribute of all persistent User instances.
However, I searched the docs and couldn't find an appropriate API (*
Session.expire_all* does not work, which is shown later;
*Session.expire*requires an instance).
I looked into SQLAlchemy's source, and finally found that I can iterate
over all the persistent objects via *Session.identity_map*. But this is not
documented, and do I really have to do this?
The simplified model is attached (sorry...it's still that long)
And I also wonder why the following does not work? It raises
an IntegrityError.
python test.py
>>> g0.perms.add(5)
>>> ses.expire_all()
>>> ses.commit()
Finally, thanks very much for your patient reading!
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To view this discussion on the web visit
https://groups.google.com/d/msg/sqlalchemy/-/tJCdWyGW2nQJ.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
from sqlalchemy import Column, event
from sqlalchemy.types import Integer, Text
from sqlalchemy.schema import ForeignKey
from sqlalchemy.orm import relationship, backref
from sqlalchemy.orm.session import object_session
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class MapUserGrpAndGrpPerm(Base):
__tablename__ = 'map_ugrp_grpperm'
def __init__(self, perm):
self.perm = perm
gid = Column(Integer, ForeignKey('ugrp.id'), primary_key = True)
perm = Column(Integer, primary_key = True)
class MapUserAndUserGrp(Base):
__tablename__ = 'map_user_ugrp'
uid = Column(Integer, ForeignKey('user.id'), primary_key = True)
gid = Column(Integer, ForeignKey('ugrp.id'), primary_key = True)
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key = True)
groups = relationship('UserGrp', secondary = 'map_user_ugrp')
_perms_cache_rst = None
_perms_cache = Column('permscache', Text())
@property
def perms(self):
if self._perms_cache_rst is None:
self._perms_cache_rst = self._get_perms()
return self._perms_cache_rst
def _get_perms(self):
if self._perms_cache is None:
rst = set()
for i in self.groups:
rst.update(i.perms)
rst = frozenset(rst)
self._perms_cache = '|' . join([str(i) for i in rst])
return rst
return frozenset([int(i) for i in self._perms_cache.split('|')])
class UserGrp(Base):
__tablename__ = 'ugrp'
id = Column(Integer, primary_key = True)
_perms = relationship('MapUserGrpAndGrpPerm', collection_class = set)
perms = association_proxy('_perms', 'perm')
def invalidate_user_perm_cache(session, gid):
for (cls, pk), obj in session.identity_map.iteritems():
if cls is User:
session.expire(obj, ['_perms_cache'])
obj._perms_cache_rst = None
sub = session.query(MapUserAndUserGrp.uid) \
.filter(MapUserAndUserGrp.gid == gid)
session.query(User).filter(User.id.in_(sub)) \
.update({User._perms_cache: None}, synchronize_session = False)
def _invcache_on_grp_perm_chg(target, *args):
ses = object_session(target)
if ses is not None:
invalidate_user_perm_cache(ses, target.id)
#for i in 'append', 'remove', 'set':
#event.listen(UserGrp._perms, i, _invcache_on_grp_perm_chg)
if __name__ == '__main__':
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine('sqlite:///:memory:', echo = True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind = engine)
ses = Session()
g0 = UserGrp()
g1 = UserGrp()
u0 = User()
u0.groups.append(g0)
u0.groups.append(g1)
g0.perms.update([1, 2])
g1.perms.update([2, 3])
ses.add(u0)
ses.commit()
import code
code.interact(local = locals())
[sqlalchemy] HOW TO HANDLE SPECIAL CHARACTERS WITH ENGINE.EXECUTE
Hello all,
I have an interesting problem for which I am sure some simple solution
must be existing.
I have made a Python function which I will paist below.
Basically what the function does is that it takes 3 parameters, namely
the name of a stored procedure, engine instance and a set of parameters.
This function is used to make calls to postgresql based stored
procedures in a modular way.
The function is kept central and all my modules just pass the necessary
parameters and leave it to the function to do the rest.
I got it working perfectly, except that I don't know how to handle
special characters when constructing the query that makes a call to a
stored procedure. So if I have an insert query which has a value with a
single quote ('), it crashes.
As you will observe in the function, it takes the arguements from a list
called queryParams and constructs that part of the query that takes
input arguements for inserts or for the where clause during select etc.
So in those input parameters if any special character appears, the said
stored procedure naturally fails.
following is the exact function.
Some one please point me out what the fundamental mistake is and
probably send in a corrected version.
def execproc(procname, engine, queryParams=[]):
""" Purpose:
executes a named stored procedure and returns the result.
Function takes 3 parameters, procname is a string containing the
name of the stored procedure.
engine is the sqlalchemy engine instance through which the query
will be executed.
queryParams contains the input parameters in a list form (if any).
description:
First it starts building a query string that commonly begins with
the common select * from syntax that is needed for calling a stored
procedure.
The code then goes to check if one or more parameters are
supplied. If yes then
a for loops runs that concatinate the parameters inside ()
During this process it checks the datatype of each supplied
parameter to stringify any parameter or keep it as integer.
This is done using the %s, %d and %f place holders.
After the query is built using the user input that consisted of the
proc name and parames, it executes the same using the supplied engine
instance.
The result of the execution contains the rows returned by the
stored procedure that was called.
"""
listCounter = 0
if len(queryParams) == 0:
queryString = "select * from %s() " % (procname)
else:
queryString = "select * from %s(" % (procname)
for param in queryParams:
if type(param) == str:
queryString = queryString + "'%s'" % (param)
if type(param) == int:
queryString = queryString + "%d" % (param)
if type(param) == float:
queryString = queryString + "%.2f" % (param)
if type(param) == NoneType:
queryString = queryString + "None"
if listCounter < (len(queryParams) - 1):
queryString = queryString + ","
listCounter = listCounter + 1
queryString = queryString + ")"
print queryString
res =
engine.execute(text(queryString).execution_options(autocommit=True))
return res
Thanks for help in advance.
Happy hacking.
Krishnakant.
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
Re: [sqlalchemy] user defined type
Just for the archive, On 08/02/2012 16:47, Michael Bayer wrote: @hybrid_property def namesandvar(self): nVar = self.name if self.name2: nVar += ', ' + self.name2 if self.variety: nVar += ', ' + self.variety return nVar I must be doing something wrong when defining my "NAMES40" custom type as when I change it to use Unicode the query works fine. those + signs should probably be concatenation operators.You'll get concatenation as long as the type of column includes "types.Concatenable" in its inheritance hierarchy. Went "back" to using straight sa.Unicode instead of my customtype and changed the hybrid_property to the following: @hybrid_property def namesandvar(self): if not self.variety in [None, u'']: return self.drinknames + u", " + self.variety else: return self.drinknames @namesandvar.expression def namesandvar(cls): return case([ (cls.variety != None, cls.drinknames + u", " + cls.variety), ], else_ = cls.drinknames) Which is based on what I found in the doc here: http://docs.sqlalchemy.org/en/latest/orm/mapper_config.html#mapper-sql-expressions Werner -- You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To post to this group, send email to sqlalchemy@googlegroups.com. To unsubscribe from this group, send email to sqlalchemy+unsubscr...@googlegroups.com. For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en.
[sqlalchemy] Delete failing with StaleDataError
The attached script fails with this:
Traceback (most recent call last):
File "stale_delete.py", line 33, in
session.flush()
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py",
line 1559, in flush
self._flush(objects)
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py",
line 1630, in _flush
flush_context.execute()
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/unitofwork.py",
line 331, in execute
rec.execute(self)
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/unitofwork.py",
line 498, in execute
uow
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/mapper.py", line
2507, in _delete_obj
(table.description, len(del_objects), c.rowcount)
sqlalchemy.orm.exc.StaleDataError: DELETE statement on table 'A' expected to
delete 2 row(s); 1 were matched.
in MySQL-InnoDB (works in SQLite and Postgres).
Tried versions 0.7.3 and 0.7.5
Python 2.7
I tried manually deleting the rows from the table through MySQL client and
noticed that it's not returning an accurate row count (doesn't
'supports_sane_rowcount') for a table with an adjacency relationship and an
ondelete='CASCADE'.
I'm going to work around this by adding another ondelete='CASCADE' for User->A
--
Fayaz Yusuf Khan
Cloud developer and architect
Dexetra SS, Bangalore, India
fayaz.yusuf.khan_AT_gmail_DOT_com
fayaz_AT_dexetra_DOT_com
+91-9746-830-823
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, backref
Base = declarative_base(bind=create_engine('mysql://root@localhost/test'))
class User(Base):
__tablename__ = 'User'
id = Column(Integer, primary_key=True)
class A(Base):
__tablename__ = 'A'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey(User.id))
user = relationship(User, backref=backref(
'a', cascade='all, delete, delete-orphan'))
parent_id = Column(Integer, ForeignKey('A.id', ondelete='CASCADE'))
Base.metadata.create_all()
session = sessionmaker()()
user = User()
session.add(user)
session.add_all([A(id=1, parent_id=1, user=user),
A(id=2, parent_id=1, user=user)])
session.flush()
session.delete(user)
session.flush()
signature.asc
Description: This is a digitally signed message part.
