Das deutsche Python-Forum

Code: Alles auswählen

[2022-12-19 18:21:12] Connected
crawler> INSERT INTO workers (worker_id, jobs_completed) VALUES ('asdasd', 4) ON CONFLICT (worker_id) DO UPDATE SET jobs_completed = (excluded.jobs_completed + workers.jobs_completed), time_updated = now()
[2022-12-19 18:21:12] 1 row affected in 149 ms

Code: Alles auswählen

from sqlalchemy import Column, DateTime, Integer, String, asc, desc, func
from sqlalchemy.exc import NoResultFound

from data_db.db_connection import async_db_session
from sqlalchemy.dialects.postgresql import insert

Base = declarative_base()

class Workers(Base, ModelAdmin):
    __tablename__ = "workers"

    id = Column(Integer, primary_key=True)
    worker_id = Column(String, nullable=False, unique=True)
    jobs_completed = Column(Integer, default=0)

    time_created = Column(DateTime(timezone=True), server_default=func.now())
    time_updated = Column(DateTime(timezone=True), onupdate=func.now())

    def __repr__(self):
        return (
            f"<{self.__class__.__name__}("
            f"id={self.id}, "
            f"worker_id={self.worker_id}, "
            f"jobs_completed={self.jobs_completed}, "
            f"time_created={self.time_created}, "
            f"time_updated={self.time_updated}, "
        )

    @classmethod
    async def add_completed_jobs_counter(cls, worker_job_list):
        #  https://docs.sqlalchemy.org/en/14/dialects/postgresql.html#insert-on-conflict-upsert
       """
              INSERT INTO workers (worker_id, jobs_completed)
              VALUES ($1, $2::INTEGER)
              ON CONFLICT (worker_id) DO UPDATE
              SET jobs_completed = (EXCLUDED.jobs_completed + workers.jobs_completed);
        """
        for worker_job in worker_job_list:
            query = insert(cls).values(dict(worker_id=worker_job[0], jobs_completed=worker_job[1]))
            on_conflict = query.on_conflict_do_update(
               index_elements=[cls.worker_id],
               set_={"jobs_completed": query.excluded.jobs_completed + cls.jobs_completed, "time_updated": func.now()}, )
            await async_db_session.execute(on_conflict)
            await async_db_session.commit()