Short Description: I want to add insertion_date and last_update columns to a table in a Postgresql database using SQLalchemy ORM. I followed the documentation and the examples on SO, but it didn't work for me.
Here is my example:
base.py
from sqlalchemy.ext.declarative.api import declarative_base
Base = declarative_base()
movie_model.py
from __future__ import annotations
import datetime
from sqlalchemy import func
from base import Base
from sqlalchemy.sql.schema import Column
from typing import Optional, Dict, Union
from sqlalchemy.sql.sqltypes import String, Date, DateTime, Integer
class Movie(Base):
__tablename__ = 'movies'
movie_id = Column(String, primary_key=True)
movie_name = Column(String)
release_date = Column(Date)
movie_revenue = Column(Integer)
last_update = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
insertion_date = Column(DateTime, server_default=func.now())
# __mapper_args__ = {"eager_defaults": True}
@staticmethod
def from_dict(movie_dict: Dict[str, Union[str, int, datetime]]) -> Optional[Movie]:
if not movie_dict:
return None
return Movie(
movie_id=movie_dict.get('movie_id'),
movie_name=movie_dict.get('movie_name'),
release_date=movie_dict.get('release_date'),
movie_revenue=movie_dict.get('movie_revenue')
)
db_manager.py
from movie_model import Movie
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.inspection import inspect
from typing import List
class DBManager:
def __init__(self, session):
self.session = session
self.columns = Movie.__table__.columns.keys()
self.primary_key = inspect(Movie).primary_key[0]
self.primary_key_name = self.primary_key.name
def add_many_on_conflict_do_update(self, movies: List[Movie]) -> None:
# Used self.columns[:-2] to exclude the last two columns (last_update and insertion_date) from the explicit insertion, so the ORM handles it
# on its own as mentioned here: https://docs.sqlalchemy.org/en/14/core/metadata.html#sqlalchemy.schema.Column.params.onupdate.
statement = insert(Movie.__table__).values([{attr: getattr(movie, attr) for attr in self.columns[:-2]} for movie in movies])
statement = statement.on_conflict_do_update(
index_elements=[self.primary_key_name], set_={attr: getattr(statement.excluded, attr) for attr in self.columns[:-2]}
)
self.session.execute(statement)
self.session.commit()
def get_all(self) -> List[Movie]:
return self.session.query(Movie).all()
main.py
from movie_model import Movie
from db_manager import DBManager
import datetime
from base import Base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
def main():
db_string = "postgresql+psycopg2://db_user:password@localhost:5432/db_name"
engine = create_engine(db_string, echo=False)
session = scoped_session(sessionmaker(bind=engine))
Base.metadata.create_all(engine, checkfirst=True)
movie1 = {
'movie_id': '1',
'movie_name': 'movie1',
'release_date': datetime.datetime.strptime('2010-09-23', '%Y-%m-%d').date(),
'movie_revenue': 1000
}
movie2 = {
'movie_id': '2',
'movie_name': 'movie2',
'release_date': datetime.datetime.strptime('2010-09-23', '%Y-%m-%d').date(),
'movie_revenue': 2000
}
movie3 = {
'movie_id': '3',
'movie_name': 'movie3',
'release_date': datetime.datetime.strptime('2010-09-24', '%Y-%m-%d').date(),
'movie_revenue': 3000
}
movie4 = {
'movie_id': '4',
'movie_name': 'movie4',
'release_date': datetime.datetime.strptime('2010-09-24', '%Y-%m-%d').date(),
'movie_revenue': 4000
}
movies = [movie1, movie2, movie3, movie4]
movies_models = []
for movie in movies:
movies_models.append(Movie.from_dict(movie))
db_manager = DBManager(session=session)
db_manager.add_many_on_conflict_do_update(movies_models)
retrieved_movies = db_manager.get_all()
for movie in retrieved_movies:
print(movie.last_update)
if __name__ == '__main__':
main()
Packages:
psycopg2-binary==2.9.1
SQLAlchemy==1.3.24
**Postgres version: ** 12.
When I run python main.py the insertion_date is inserted properly even if I run it for many times adding new movie each time. But the last_update is only inserted the first time I add a movie, while it is expected to be updated whenever an update is applied to the already inserted movie.
I tried replacing self.columns[:-2] by self.columns[:-1] or by self.columns in the insertion statement and/or in the _set parameter. But it ended up that last_update column is null or it is being updated every time I run main.py even if there are no updates/insertions to apply.
I also replaced onupdate by server_onupdate and datetime.datetime.now by func.now() in movie_model.py but nothing worked.
I even added __mapper_args__ = {"eager_defaults": True} to Movie() in movie_model.py as described here but still didn't work.
I checked the following questions on SO but nothing worked too:
- Is possible to create Column in SQLAlchemy which is going to be automatically populated with time when it inserted/updated last time?
- onupdate not overridinig current datetime value
Any suggestions to make it work?
Edit for clarification:
What I am looking for is to make the last_update column get updated with the time of the transaction whenever a row is updated (using on_conflict_do_update).