[sqlalchemy] Additional error handling using MySQLdb.connection extended API
Hi, I need access to MySQLdb.connection.error() and MySQLdb.connection.errno(), as we have a database which is throwing custom error codes that are higher than CR_MAX_ERROR (for reference: https://github.com/PyMySQL/mysqlclient-python/blob/master/_mysql.c#L124) My actual code looks like something along these lines: from sqlalchemy import create_engine, text engine = create_engine(url) sql = "SELECT... " try: conn = engine.connect() rows = conn.execute(text(sql)) except Exception as e: if hasattr(conn.connection, "error"): print(conn.connection.errno()) print(conn.connection.error()) I've also tried: err_conn = conn._branch() try: err_conn = engine.connect() rows = err_conn.execute(text(sql)) except Exception as e: if hasattr(err_conn.connection, "error"): print(err_conn.connection.errno()) print(err_conn.connection.error()) with no luck. I'm not sure how to write a minimal test that just uses vanilla MySQL, because I'm not sure how to trigger an error with a errno > CR_MAX_ERROR other than in our system. I'm losing the information somehow, via a connection being closed/reopened or something? I'm not really sure. What might be the a way to handle this without just using raw_connection or writing a new dialect? Thanks, Brian -- 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 sqlalchemy+unsubscr...@googlegroups.com. To post to this group, send email to sqlalchemy@googlegroups.com. Visit this group at https://groups.google.com/group/sqlalchemy. For more options, visit https://groups.google.com/d/optout.
[sqlalchemy] Advice for Implementing a ShortUUID Column Type?
I want to make my UUID's prettier so I've gone about implementing a ShortUUID column based on the shortuuid library[1]. The idea is to store the primary key as a UUID type in postgres (since its optimized for that) and transform the UUID to a shortuuid for presentation and querying. This is my first attempt at implementing it. It has some short comings. I was wondering if you had any advice for fully baking the implementation. I've pasted the code I have so far. from sqlalchemy_utils.types.uuid import UUIDType import uuid import shortuuid def _decode_shortuuid(value): try: return shortuuid.decode(value) except ValueError: return None def _encode_shortuuid(value): try: if value is None: return None return shortuuid.encode(value) except KeyError: return None class ShortUUID(UUIDType): """Converts UUIDs to ShortUUIDs for readability's sake.""" def process_bind_param(self, value, dialect): """Process a ShortUUID to a UUID.""" if value is None: return value if type(value) != uuid.UUID: value = _decode_shortuuid(value) return super().process_bind_param(value, dialect) def process_result_value(self, value, dialect): """Return a ShortUUID encoded UUID.""" value = super().process_result_value(value, dialect) return _encode_shortuuid(value) 1. https://github.com/skorokithakis/shortuuid -- 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 sqlalchemy+unsubscr...@googlegroups.com. To post to this group, send email to sqlalchemy@googlegroups.com. Visit this group at https://groups.google.com/group/sqlalchemy. For more options, visit https://groups.google.com/d/optout.
[sqlalchemy] Purpose of assertion in table reflection test (1.1.9)
Upgrading a specific dialect (EXASOL) to 1.1.9 made me stumble across this
test (part of test_reflection.py):
https://github.com/zzzeek/sqlalchemy/blob/rel_1_1_9/lib/sqlalchemy/testing/suite/test_reflection.py#L737
def test_get_table_names(self):
tablenames = [
t for t in inspect(testing.db).get_table_names()
if t.lower() in ("t1", "t2")]
eq_(tablenames[0].upper(), tablenames[0].lower())
eq_(tablenames[1].upper(), tablenames[1].lower())
The test case fails and this does not look too unexpected to me. The tables
T1 and T2 are added to the tablenames array, and u"t1".upper() is always
different from u"t1".lower().
Am I missing the purpose of the test case or is this always bound to fail?
--
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 sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
Re: [sqlalchemy] Re: Generic "soft delete" implementation
On 05/18/2017 09:20 AM, Jaco Breitenbach wrote:
Hi Mike,
Thank you very much. I've managed to get it working after a fashion.
Unfortunately this technique seems to interfere with the automatic
clean-up of the secondary (mapping) table in the case of a many-to-many
relationship. In the example below I have a contact_group table and
user table and a many-to-many relationship set up via a
contact_group_user_map table. Since I've used backref in defining the
relationship, I expect the row in contact_group_user_map to
automatically be deleted when I do session.delete(userObj). This works
perfectly as long as the before_flush handler is disabled. If I
intercept the delete of userObj and turn it into an update, the delete
from contact_group_user_map no longer happens.
well you are turning delete into update, so if you also want to mutate
that collection, you need to do that manually. should be easy w/ the
backref, e.g.:
my_soft_deleted_instance.users = []
I've noticed that the
list of objects in session.deleted which is passed to the before_flush
handler does not include the contact_group_user_map entity, so I guess
the determination to delete the row from contact_group_user_map takes
place after the before_flush handler returns. Do you have any advice on
resolving this issue?
from sqlalchemy import create_engine, event, BigInteger, Column,
DateTime, Float, ForeignKey, Index, Integer, String, text
from sqlalchemy.orm import attributes, relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy_repr import PrettyRepresentableBase
Base = declarative_base(cls=PrettyRepresentableBase)
metadata = Base.metadata
engine = None
Session = None
def handle__before_flush(session, flush_context, instances):
print("handle__before_flush()")
for instance in session.deleted:
print(instance)
if not attributes.instance_state(instance).has_identity:
continue
if not hasattr(instance, '_deleted'):
continue
#instance._deleted = True
#session.add(instance)
def getSession(conn):
global engine, Session
if engine is None or Session is None:
engine = create_engine(conn, echo=True)
Session = sessionmaker(bind=engine, query_cls=SoftDeleteQuery)
event.listen(Session, 'before_flush', handle__before_flush)
return Session()
class ContactGroup(Base):
__tablename__ = 'contact_group'
id = Column(BigInteger, primary_key=True)
_deleted = Column(Integer, nullable=False, server_default=text("'0'"))
name = Column(String(200), nullable=False)
description = Column(String(1000))
users = relationship("User", secondary="contact_group_user_map",
backref="contact_groups")
class ContactGroupUserMap(Base):
__tablename__ = 'contact_group_user_map'
id = Column(BigInteger, primary_key=True)
contact_group_id = Column(ForeignKey(u'contact_group.id'),
nullable=False, index=True)
user_id = Column(ForeignKey(u'user.id'), nullable=False, index=True)
contact_group = relationship(u'ContactGroup')
user = relationship(u'User')
class User(Base):
__tablename__ = 'user'
id = Column(BigInteger, primary_key=True)
_deleted = Column(Integer, nullable=False, server_default=text("'0'"))
username = Column(String(200), nullable=False, unique=True)
password = Column(String(1000))
Thank you and kind regards,
Jaco
On Wednesday, 17 May 2017 23:30:25 UTC+1, Mike Bayer wrote:
On 05/17/2017 05:25 PM, Jaco Breitenbach wrote:
> Thank you, Jonathan, that wiki example has helped a lot. I've
extended
> it a bit to also add the extra condition to joined tables.
>
> I want to take it one step further now and automatically convert
> session.delete()s into update statements (set is_deleted = true)
instead
> of delete statements. I've experimented with a custom
SessionExtension,
don't use "Extensions", use the event system:
http://docs.sqlalchemy.org/en/rel_1_1/orm/events.html?highlight=before_flush#sqlalchemy.orm.events.SessionEvents.before_flush
> session.deleted.pop()# Remove the instance
from the
> list of entities to be deleted
ouch...that's not going to do anything. everything in Python is
mutable and it's kind of hard to make every collection everywhere that
only is a view of things immutable (though this is custom collection,
could be made immutable with a flag or something). To undo the
delete,
add the object back, session.add(object).
>
>
> Thank you and kind regards,
> Jaco
>
>
> On Tuesday, 16 May 2017 17:20:12 UTC+1, Jonathan Vanasco wrote:
>
> The FAQ
>
Re: [sqlalchemy] Re: Generic "soft delete" implementation
Hi Mike,
Thank you very much. I've managed to get it working after a fashion.
Unfortunately this technique seems to interfere with the automatic
clean-up of the secondary (mapping) table in the case of a many-to-many
relationship. In the example below I have a contact_group table and user
table and a many-to-many relationship set up via a contact_group_user_map
table. Since I've used backref in defining the relationship, I expect the
row in contact_group_user_map to automatically be deleted when I do
session.delete(userObj). This works perfectly as long as the before_flush
handler is disabled. If I intercept the delete of userObj and turn it into
an update, the delete from contact_group_user_map no longer happens. I've
noticed that the list of objects in session.deleted which is passed to the
before_flush handler does not include the contact_group_user_map entity, so
I guess the determination to delete the row from contact_group_user_map
takes place after the before_flush handler returns. Do you have any advice
on resolving this issue?
from sqlalchemy import create_engine, event, BigInteger, Column, DateTime,
Float, ForeignKey, Index, Integer, String, text
from sqlalchemy.orm import attributes, relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy_repr import PrettyRepresentableBase
Base = declarative_base(cls=PrettyRepresentableBase)
metadata = Base.metadata
engine = None
Session = None
def handle__before_flush(session, flush_context, instances):
print("handle__before_flush()")
for instance in session.deleted:
print(instance)
if not attributes.instance_state(instance).has_identity:
continue
if not hasattr(instance, '_deleted'):
continue
#instance._deleted = True
#session.add(instance)
def getSession(conn):
global engine, Session
if engine is None or Session is None:
engine = create_engine(conn, echo=True)
Session = sessionmaker(bind=engine, query_cls=SoftDeleteQuery)
event.listen(Session, 'before_flush', handle__before_flush)
return Session()
class ContactGroup(Base):
__tablename__ = 'contact_group'
id = Column(BigInteger, primary_key=True)
_deleted = Column(Integer, nullable=False, server_default=text("'0'"))
name = Column(String(200), nullable=False)
description = Column(String(1000))
users = relationship("User", secondary="contact_group_user_map",
backref="contact_groups")
class ContactGroupUserMap(Base):
__tablename__ = 'contact_group_user_map'
id = Column(BigInteger, primary_key=True)
contact_group_id = Column(ForeignKey(u'contact_group.id'),
nullable=False, index=True)
user_id = Column(ForeignKey(u'user.id'), nullable=False, index=True)
contact_group = relationship(u'ContactGroup')
user = relationship(u'User')
class User(Base):
__tablename__ = 'user'
id = Column(BigInteger, primary_key=True)
_deleted = Column(Integer, nullable=False, server_default=text("'0'"))
username = Column(String(200), nullable=False, unique=True)
password = Column(String(1000))
Thank you and kind regards,
Jaco
On Wednesday, 17 May 2017 23:30:25 UTC+1, Mike Bayer wrote:
>
>
>
> On 05/17/2017 05:25 PM, Jaco Breitenbach wrote:
> > Thank you, Jonathan, that wiki example has helped a lot. I've extended
> > it a bit to also add the extra condition to joined tables.
> >
> > I want to take it one step further now and automatically convert
> > session.delete()s into update statements (set is_deleted = true) instead
> > of delete statements. I've experimented with a custom SessionExtension,
>
> don't use "Extensions", use the event system:
>
>
> http://docs.sqlalchemy.org/en/rel_1_1/orm/events.html?highlight=before_flush#sqlalchemy.orm.events.SessionEvents.before_flush
>
>
>
>
> > session.deleted.pop()# Remove the instance from the
> > list of entities to be deleted
>
> ouch...that's not going to do anything. everything in Python is
> mutable and it's kind of hard to make every collection everywhere that
> only is a view of things immutable (though this is custom collection,
> could be made immutable with a flag or something). To undo the delete,
> add the object back, session.add(object).
>
>
>
>
>
>
>
> >
> >
> > Thank you and kind regards,
> > Jaco
> >
> >
> > On Tuesday, 16 May 2017 17:20:12 UTC+1, Jonathan Vanasco wrote:
> >
> > The FAQ
> >
> http://docs.sqlalchemy.org/en/latest/faq/sessions.html#how-do-i-make-a-query-that-always-adds-a-certain-filter-to-every-query
>
> > <
> http://docs.sqlalchemy.org/en/latest/faq/sessions.html#how-do-i-make-a-query-that-always-adds-a-certain-filter-to-every-query>
>
>
> >
> > points to an example in the wiki:
> >
> https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/PreFilteredQuery
> > <
>
