I have been using MSSQL
and pyodbc
and updating a column with DateTime
type by following the below thread:
How to update datetime field in MSSQL using python pyodbc module
Now I am also trying to incorporate sqlalchemy
in my application stack. I am aware of following answer but it does not serve my purpose.
Datetime not updating on insert using SQLAlchemy on MSSQL
To elaborate on the problem:
from sqlalchemy import Column, String, DateTime, Date
from sqlalchemy.ext.declarative import declarative_base
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from datetime import datetime as datetimemodule, date as datemodule
import traceback
BASE = declarative_base()
class REPR:
def make_repr(self, **kwrgs):
argument_string = ', '.join([f'{a}={b}' for a, b in locals()['kwrgs'].items()])
display_string = f'{self.__class__.__name__}({argument_string})'
return '*****' + display_string + '*****'
def __str__(self):
return self.__repr__()
class Recog_EntryLog(BASE, REPR):
# In database corresponding table has no primary key.
# As suggested in the documentation, largest candidate key is marked as
# primary key here.
__tablename__ = 'test'
id = Column(String(100), primary_key=True)
entrytime = Column(DateTime, primary_key=True) # I want to update this field by custom defined time.
# Not by the entry update time.
entrydate = Column(Date, primary_key=True)
def __repr__(self):
return self.make_repr(id=self.id, entrytime=self.entrytime, entrydate=self.entrydate)
def get_connection(dbname):
connection_info = f'mssql+pyodbc://remote:prashant@127.0.0.1:1433/{dbname}?driver=ODBC+Driver+17+for+SQL+Server'
engine = create_engine(connection_info, echo=True)
session = sessionmaker(bind=engine)()
return session, engine
def close_connection(connection_info:sqlalchemy.orm.session.Session,
engine:sqlalchemy.engine.base.Engine):
connection_info.close()
engine.dispose()
def search_and_update(connected_session:sqlalchemy.orm.session.Session,
id,
_time:datetimemodule,
_date:datemodule):
res = connected_session.query(Recog_EntryLog).filter_by(id=id, entrydate=_date).order_by(Recog_EntryLog.entrytime.desc()).first()
print(res)
try:
if res is None:
connected_session.add(Recog_EntryLog(id=id, entrytime=_time, entrydate=_date))
else:
query = connected_session.query(Recog_EntryLog).filter_by(id=id, entrytime=res.entrytime, entrydate=_date)
query.update({Recog_EntryLog.entrytime: _time})
connected_session.commit()
except:
traceback.print_exc()
connected_session.rollback()
if __name__ == '__main__':
connected_session, connected_engine = get_connection('authentication')
# Will create new entry on first run/Will update old entry
search_and_update(connected_session, '1234', datetimemodule.now(), datemodule.today())
# should update datetime
search_and_update(connected_session, '1234', datetimemodule.now(), datemodule.today())
# should update datetime
search_and_update(connected_session, '1234', datetimemodule.now(), datemodule.today())
close_connection(connected_session, connected_engine)
I was expecting that last two red values will be updated with the first two blue values. Generated SQL does not seem to follow first attach link sql.
Can some help me how to properly write such function using sqlalchemy?