feel free to hang here, note that some of the "chat" happens on libera.chat #sqlalchemy
Hi all. I have a column (postgresql 14) that I thought would update on every row update (1.4.27), declared as follows:
class utcnow(expression.FunctionElement): # pylint: disable=W0223
type = DateTime()
...
updated_at = Column(
DateTime(True),
nullable=False,
onupdate=utcnow(),
server_default=utcnow(),
index=True,
)This was what I came up with. I had the impression it works... sometimes. What am I missing?
1 reply
Hello,
I have a problem but I'm not sure whether it belongs here or with sqlalchemy. Once that determined I'll provide the required details.
Using sqlalchemy's PickleType I'm loading large files BLOBs into Oracle. Unfortunatelly they are too large and I get DPI-1057: buffer size too large.
From oracle/python-cx_Oracle#480 I learned that streaming writes must be used in order to write LOBs larger than 1073741822B.
As far as I know cx-Oracle provides DB-API implementation that's used by sqlalchemy, but I'm uncertain on exactly where is the boundary.
This brings me to my question:
Does support for streaming writes (to support large LOBs) is something to be implemented in python-cx-Oracle or in sqlalchemy?
@CaselIT
I think I'm stuck because of not understanding the difference between different abstractions:
- cx-Oracle
- DBAPI
- sqlalchemy dialects
- sqlalchemy core
My current understanding is that cx-Oracle is Python wrapper to Oracle Client libraries. It exposes a DBAPI (PEP249) compatibile interface
I'm unclear about "Oracle dialect" - https://docs.sqlalchemy.org/en/14/dialects/.
Is that a separate thing that sits between core and the DBAPI driver? Or it's part of either cx-Oracle or SQLAlchemy Core?
That sounds like a reasonable approach. We're using SQLAlchemy across the whole application, so it'd be nice to be able to interface is somehow.
Would it go something like this?
connection = engine.raw_connection()
cursor = connection.cursor()https://stackoverflow.com/a/49816320/12640922
And then that's the same cursor & connection they are using here:
id_val = 9
lob_var = cursor.var(cx_Oracle.DB_TYPE_BLOB)
cursor.execute("""
insert into lob_tbl (id, b)
values (:1, empty_blob())
returning b into :2""", [id_val, lob_var])
blob, = lobVar.getvalue()
offset = 1
num_bytes_in_chunk = 65536
with open("image.png", "rb") as f:
while True:
data = f.read(num_bytes_in_chunk)
if data:
blob.write(data, offset)
if len(data) < num_bytes_in_chunk:
break
offset += len(data)
connection.commit()https://cx-oracle.readthedocs.io/en/latest/user_guide/lob_data.html#streaming-lobs-write
class Box(BaseModel):
__tablename__ = "box"
id = Column(Integer, primary_key=True)
category = Column(String)
class Folder(BaseModel):
__tablename__ = "folder"
id = Column(Integer, primary_key=True)
box_id = Column(Integer, ForeignKey('box.id'))
box = relationship(Box, backref=backref("folders"))
category = ??? # In particular, I don't want this column to ACTUALLY exist in this table
# I'd like:
session.query(Folder).filter(Folder.category == "Mid-17th-Century Pocket Lint Collectors")
# to implicitly result in SQL with the effect of:
SELECT * FROM folder, box WHERE folder.box_id = box.id and box.category = 'Mid-17th-Century Pocket Lint Collectors'category = column_property(box.category) should work, but apparently the box field isn't populated yet at that point, because I get an AttributeError on box.category.
@CaselIT: So using association proxy with my earlier example, like
class Folder(BaseModel):
...
box_id = Column(Integer, ForeignKey('box.id'))
box = relationship(Box, backref=backref("folders"))
category = association_proxy("box", "category")Though referencing it works in some contexts (like .filter(Folder.category == "x"), it doesn't transparently work in other field contexts. Specifically:
session.query(Folder.category).all()
-> InvalidRequestError: SQL expression, column, or mapped entity expected - got 'ColumnAssociationProxyInstance(AssociationProxy('box', 'category'))'Any way to make that work? What I want is for callers not to have to know that this isn't a real column.
SELECT associated_table.associated_col_name AS col_name ...?
select specifically - I just meant the SELECT expressions in the compiled SQL resulting from session.query(Model.an_association_proxy_attr).
column_property() is that in my actual usage, I have like a dozen such mapped fields, and the result is that the generated query has like a dozen subselects to the same related table. I would kinda like the eager loading, but not at that expense. :P
association_proxy doesn't work with order_by(), either. While using it in select expressions was a nice-to-have, order_by actually is a show-stopper for me. sigh
@CaselIT: So I came up with this:
def association_hybrid_proxy(target: str, attr: str):
prop = hybrid_property(lambda self: getattr(getattr(self, target), attr))
return prop.expression(lambda cls: getattr(getattr(cls, target).mapper.class_, attr))You use it just like association_proxy, but it makes a hybrid_property under the covers that works in both filter() and order_by() (the latter requires that you ensure the join in some other way, so it's not 100% transparent, but at least that's easy and immediately discovered if you don't).
declared_attr. I'll get around to it in a bit and know for sure.
:point_up: Edit: @CaselIT: So I came up with this:
def hybrid_association_proxy(target: str, attr: str):
prop = hybrid_property(lambda self: getattr(getattr(self, target), attr))
return prop.expression(lambda cls: getattr(getattr(cls, target).mapper.class_, attr))You use it just like association_proxy, but it makes a hybrid_property under the covers that works in both filter() and order_by() (the latter requires that you ensure the join in some other way, so it's not 100% transparent, but at least that's easy and immediately discovered if you don't).
import pandas as pd
import snowflake.connector as snow
from snowflake.connector.pandas_tools import write_pandas
import logging
SNOWFLAKE_ACCOUNT = 'jwa13025' # os.getenv('SF_ACOUNT') SF_ACOUNT
SNOWFLAKE_USER = 'nkumar' # os.getenv('SF_USR') SF_USR
SNOWFLAKE_PWD = 'Kuldeep@12' # os.getenv('SF_PWD') SF_PWD'
conn = snow.connect(user=SNOWFLAKE_USER,
password=SNOWFLAKE_PWD ,
account=SNOWFLAKE_ACCOUNT,
warehouse="WORLDPANTRYDW",
database="SDS",
schema="PUBLIC");
try:
logging.basicConfig(level=logging.INFO, filename='Load_SDS_Header.log', filemode='w', format='%(asctime)s %(message)s',datefmt='%m/%d/%Y %I:%M:%S %p')
Result=conn.cursor().execute("use schema PUBLIC")
print(Result)
SDS_HEADERS = pd.read_csv("C:\BalaReports\SDS\Oracle_SDS_Datafiles_Nov21\Headers.txt",sep="\t",header=None,names=["Column_Id","Header_ColId","Header_ColDescription"])
print(Header_Info)
success, nchunks, nrows, _ = write_pandas(conn, SDS_HEADERS, "SDS.PUBLIC.SDS_HEADERS",schema="PUBLIC")
print(success)except conn.error.DatabaseError as e:
#raise e.errno
logging.error("Exception occurred", exc_info=True)except exception as e:
#raise e.errno
#print('SQL Execution Error: {0}'.format(e.msg))
#print('Snowflake Query Id: {0}'.format(e.sfqid))
#print('Error Number: {0}'.format(e.errno))
#print('SQL State: {0}'.format(e.sqlstate))finally:
conn.close()
logging.info('Trace Other Issue', exc_info=True)
Object does not exist, or operation cannot be performed.
"'''"import pandas as pd"'''"
"'''"import snowflake.connector as snow "'''"
"'''"from snowflake.connector.pandas_tools import write_pandas"'''"
"'''"import logging"'''"
"'''"conn = snow.connect(user=SNOWFLAKE_USER,"'''"
"'''" password=SNOWFLAKE_PWD ,"'''"
"'''" account=SNOWFLAKE_ACCOUNT,"'''"
"'''" warehouse="WORLDPANTRYDW", "'''"
"'''" database="SDS", "'''"
"'''" schema="PUBLIC"); "'''"
"'''"try: "'''"
"'''" logging.basicConfig(level=logging.INFO, filename='Load_SDS_Header.log', filemode='w', format='%(asctime)s %(message)s',datefmt='%m/%d/%Y %I:%M:%S %p') "'''"
"'''" Result=conn.cursor().execute("use schema PUBLIC") "'''"
"'''" print(Result) "'''"
"'''" SDS_HEADERS = pd.read_csv("C:\BalaReports\SDS\Oracle_SDS_Datafiles_Nov21\Headers.txt",sep="\t",header=None,names=["Column_Id","Header_ColId","Header_ColDescription"]) "'''"
"'''" #print(SDSHEADERS) "'''"
"'''" success, nchunks, nrows, = write_pandas(conn, SDS_HEADERS, "SDS.PUBLIC.SDS_HEADERS",schema="PUBLIC") "'''"
"'''" print(success) "'''"
"'''"#except conn.error.DatabaseError as e: "'''"
"'''" #raise e.errno "'''"
"'''" logging.error("Exception occurred", exc_info=True) "'''"
"'''"#except exception as e: "'''"
"'''" #raise e.errno "'''"
"'''" #print('SQL Execution Error: {0}'.format(e.msg)) "'''"
"'''" #print('Snowflake Query Id: {0}'.format(e.sfqid)) "'''"
"'''" #print('Error Number: {0}'.format(e.errno)) "'''"
"'''" #print('SQL State: {0}'.format(e.sqlstate)) "'''"
"'''"finally: "'''"
"'''" conn.close() "'''"
"'''" logging.info('Trace Other Issue', exc_info=True) "'''"