def process_entities(self, tlo: Any) -> None:
        rows = self._entities_to_rows(tlo)
        if not rows:
            return

        t = self.Entity.__table__
        ins = insert(t)
        upsert = ins.on_conflict_do_update(constraint=t.primary_key, set_={
            "hash": ins.excluded.hash,
            "username": ins.excluded.username,
            "phone": ins.excluded.phone,
            "name": ins.excluded.name,
        })
        with self.engine.begin() as conn:
            conn.execute(upsert, [dict(session_id=self.session_id, id=row[0], hash=row[1],
                                       username=row[2], phone=row[3], name=row[4])
                                  for row in rows]) 

def store_scan_doc(self, scan):
        scan = scan.copy()
        if 'start' in scan:
            scan['start'] = datetime.datetime.utcfromtimestamp(
                int(scan['start'])
            )
        if 'scaninfos' in scan:
            scan["scaninfo"] = scan.pop('scaninfos')
        scan["sha256"] = utils.decode_hex(scan.pop('_id'))
        insrt = insert(self.tables.scanfile).values(
            **dict(
                (key, scan[key])
                for key in ['sha256', 'args', 'scaninfo', 'scanner', 'start',
                            'version', 'xmloutputversion']
                if key in scan
            )
        )
        if config.DEBUG:
            scanfileid = self.db.execute(
                insrt.returning(self.tables.scanfile.sha256)
            ).fetchone()[0]
            utils.LOGGER.debug("SCAN STORED: %r", utils.encode_hex(scanfileid))
        else:
            self.db.execute(insrt) 

def set_show(self, string_id):
		"""
			Set current show.
		"""
		shows = self.metadata.tables["shows"]
		with self.engine.begin() as conn:
			# need to update to get the `id`
			query = insert(shows).returning(shows.c.id)
			query = query.on_conflict_do_update(
				index_elements=[shows.c.string_id],
				set_={
					'string_id': query.excluded.string_id,
				},
			)
			self.show_id, = conn.execute(query, {
				"name": string_id,
				"string_id": string_id,
			}).first() 

def update_fact_notification_status(data, process_day, notification_type):
    table = FactNotificationStatus.__table__
    FactNotificationStatus.query.filter(
        FactNotificationStatus.bst_date == process_day,
        FactNotificationStatus.notification_type == notification_type
    ).delete()

    for row in data:
        stmt = insert(table).values(
            bst_date=process_day,
            template_id=row.template_id,
            service_id=row.service_id,
            job_id=row.job_id,
            notification_type=notification_type,
            key_type=row.key_type,
            notification_status=row.status,
            notification_count=row.notification_count,
        )
        db.session.connection().execute(stmt) 

def _insert_inbound_sms_history(subquery, query_limit=10000):
    offset = 0
    inbound_sms_query = db.session.query(
        *[x.name for x in InboundSmsHistory.__table__.c]
    ).filter(InboundSms.id.in_(subquery))
    inbound_sms_count = inbound_sms_query.count()

    while offset < inbound_sms_count:
        statement = insert(InboundSmsHistory).from_select(
            InboundSmsHistory.__table__.c,
            inbound_sms_query.limit(query_limit).offset(offset)
        )

        statement = statement.on_conflict_do_nothing(
            constraint="inbound_sms_history_pkey"
        )
        db.session.connection().execute(statement)

        offset += query_limit 

def insert_or_update_returned_letters(references):
    data = _get_notification_ids_for_references(references)
    for row in data:
        table = ReturnedLetter.__table__

        stmt = insert(table).values(
            reported_at=datetime.utcnow().date(),
            service_id=row.service_id,
            notification_id=row.id,
            created_at=datetime.utcnow()
        )

        stmt = stmt.on_conflict_do_update(
            index_elements=[table.c.notification_id],
            set_={
                'reported_at': datetime.utcnow().date(),
                'updated_at': datetime.utcnow()
            }
        )
        db.session.connection().execute(stmt) 

def dao_create_or_update_daily_sorted_letter(new_daily_sorted_letter):
    '''
    This uses the Postgres upsert to avoid race conditions when two threads try and insert
    at the same row. The excluded object refers to values that we tried to insert but were
    rejected.
    http://docs.sqlalchemy.org/en/latest/dialects/postgresql.html#insert-on-conflict-upsert
    '''
    table = DailySortedLetter.__table__
    stmt = insert(table).values(
        billing_day=new_daily_sorted_letter.billing_day,
        file_name=new_daily_sorted_letter.file_name,
        unsorted_count=new_daily_sorted_letter.unsorted_count,
        sorted_count=new_daily_sorted_letter.sorted_count)
    stmt = stmt.on_conflict_do_update(
        index_elements=[table.c.billing_day, table.c.file_name],
        set_={
            'unsorted_count': stmt.excluded.unsorted_count,
            'sorted_count': stmt.excluded.sorted_count,
            'updated_at': datetime.utcnow()
        }
    )
    db.session.connection().execute(stmt) 

def batch_upsert_records(
    session: Session, table: Table, records: List[Dict[str, Any]]
) -> None:
    """Batch upsert records into postgresql database."""
    if not records:
        return
    for record_batch in _batch_postgres_query(table, records):
        stmt = insert(table.__table__)
        stmt = stmt.on_conflict_do_update(
            constraint=table.__table__.primary_key,
            set_={
                "keys": stmt.excluded.get("keys"),
                "values": stmt.excluded.get("values"),
            },
        )
        session.execute(stmt, record_batch)
        session.commit() 

def __init__(self, db):
        self.db = db
        self.invalidated_pageids = set()

        wspc_sync = self.db.ws_parser_cache_sync
        wspc_sync_ins = insert(wspc_sync)

        self.sql_inserts = {
            "templatelinks": self.db.templatelinks.insert(),
            "pagelinks": self.db.pagelinks.insert(),
            "imagelinks": self.db.imagelinks.insert(),
            "categorylinks": self.db.categorylinks.insert(),
            "langlinks": self.db.langlinks.insert(),
            "iwlinks": self.db.iwlinks.insert(),
            "externallinks": self.db.externallinks.insert(),
            "redirect": self.db.redirect.insert(),
            "section": self.db.section.insert(),
            "ws_parser_cache_sync":
                wspc_sync_ins.on_conflict_do_update(
                    constraint=wspc_sync.primary_key,
                    set_={"wspc_rev_id": wspc_sync_ins.excluded.wspc_rev_id}
                )
        } 

def _set_sync_timestamp(self, timestamp, conn=None):
        """
        Set a last-sync timestamp for the grabber. Writes into the custom
        ``ws_sync`` table.

        :param datetime.datetime timestamp: the new timestamp
        :param conn: an existing :py:obj:`sqlalchemy.engine.Connection` or
            :py:obj:`sqlalchemy.engine.Transaction` object to be re-used for
            execution of the SQL query
        """
        ws_sync = self.db.ws_sync
        ins = insert(ws_sync)
        ins = ins.on_conflict_do_update(
                    constraint=ws_sync.primary_key,
                    set_={"wss_timestamp": ins.excluded.wss_timestamp}
                )
        entry = {
            "wss_key": self.__class__.__name__,
            "wss_timestamp": timestamp,
        }

        if conn is None:
            conn = self.db.engine.connect()
        conn.execute(ins, entry) 

def create_upsert_postgres(table, record):
    """Creates a statement for inserting the passed record to the passed
    table; if the record already exists, the existing record will be updated.
    This uses PostgreSQL `on_conflict_do_update` (hence upsert), and that
    why the returned statement is just valid for PostgreSQL tables. Refer to
    this `SqlAlchemy PostgreSQL documentation`_ for more information.

    The created statement is not executed by this function.

    Args:
        table (sqlalchemy.sql.schema.Table): database table metadata.
        record (dict): a data record, corresponding to one row, to be inserted.

    Returns:
        sqlalchemy.sql.dml.Insert: a statement for inserting the passed
        record to the specified table.

    .. _SqlAlchemy PostgreSQL documentation:
       https://docs.sqlalchemy.org/en/latest/dialects/postgresql.html#insert-on-conflict-upsert
    """
    insert_stmt = postgres_insert(table).values(record)
    return insert_stmt.on_conflict_do_update(
        index_elements=[col for col in table.primary_key],
        set_=record
    ) 

def test_bad_args(self):
        assert_raises(
            ValueError,
            insert(self.tables.users).on_conflict_do_nothing,
            constraint="id",
            index_elements=["id"],
        )
        assert_raises(
            ValueError,
            insert(self.tables.users).on_conflict_do_update,
            constraint="id",
            index_elements=["id"],
        )
        assert_raises(
            ValueError,
            insert(self.tables.users).on_conflict_do_update,
            constraint="id",
        )
        assert_raises(
            ValueError, insert(self.tables.users).on_conflict_do_update
        ) 

def test_on_conflict_do_nothing(self):
        users = self.tables.users

        with testing.db.connect() as conn:
            result = conn.execute(
                insert(users).on_conflict_do_nothing(),
                dict(id=1, name="name1"),
            )
            eq_(result.inserted_primary_key, (1,))
            eq_(result.returned_defaults, None)

            result = conn.execute(
                insert(users).on_conflict_do_nothing(),
                dict(id=1, name="name2"),
            )
            eq_(result.inserted_primary_key, (1,))
            eq_(result.returned_defaults, None)

            eq_(
                conn.execute(users.select().where(users.c.id == 1)).fetchall(),
                [(1, "name1")],
            ) 

def test_on_conflict_do_nothing_connectionless(self, connection):
        users = self.tables.users_xtra

        result = connection.execute(
            insert(users).on_conflict_do_nothing(constraint="uq_login_email"),
            dict(name="name1", login_email="email1"),
        )
        eq_(result.inserted_primary_key, (1,))
        eq_(result.returned_defaults, (1,))

        result = connection.execute(
            insert(users).on_conflict_do_nothing(constraint="uq_login_email"),
            dict(name="name2", login_email="email1"),
        )
        eq_(result.inserted_primary_key, None)
        eq_(result.returned_defaults, None)

        eq_(
            connection.execute(
                users.select().where(users.c.id == 1)
            ).fetchall(),
            [(1, "name1", "email1", None)],
        ) 

def test_on_conflict_do_update_one(self):
        users = self.tables.users

        with testing.db.connect() as conn:
            conn.execute(users.insert(), dict(id=1, name="name1"))

            i = insert(users)
            i = i.on_conflict_do_update(
                index_elements=[users.c.id], set_=dict(name=i.excluded.name)
            )
            result = conn.execute(i, dict(id=1, name="name1"))

            eq_(result.inserted_primary_key, (1,))
            eq_(result.returned_defaults, None)

            eq_(
                conn.execute(users.select().where(users.c.id == 1)).fetchall(),
                [(1, "name1")],
            ) 

def test_on_conflict_do_update_three(self):
        users = self.tables.users

        with testing.db.connect() as conn:
            conn.execute(users.insert(), dict(id=1, name="name1"))

            i = insert(users)
            i = i.on_conflict_do_update(
                index_elements=users.primary_key.columns,
                set_=dict(name=i.excluded.name),
            )
            result = conn.execute(i, dict(id=1, name="name3"))
            eq_(result.inserted_primary_key, (1,))
            eq_(result.returned_defaults, None)

            eq_(
                conn.execute(users.select().where(users.c.id == 1)).fetchall(),
                [(1, "name3")],
            ) 

def test_on_conflict_do_update_four(self):
        users = self.tables.users

        with testing.db.connect() as conn:
            conn.execute(users.insert(), dict(id=1, name="name1"))

            i = insert(users)
            i = i.on_conflict_do_update(
                index_elements=users.primary_key.columns,
                set_=dict(id=i.excluded.id, name=i.excluded.name),
            ).values(id=1, name="name4")

            result = conn.execute(i)
            eq_(result.inserted_primary_key, (1,))
            eq_(result.returned_defaults, None)

            eq_(
                conn.execute(users.select().where(users.c.id == 1)).fetchall(),
                [(1, "name4")],
            ) 

def test_on_conflict_do_update_five(self):
        users = self.tables.users

        with testing.db.connect() as conn:
            conn.execute(users.insert(), dict(id=1, name="name1"))

            i = insert(users)
            i = i.on_conflict_do_update(
                index_elements=users.primary_key.columns,
                set_=dict(id=10, name="I'm a name"),
            ).values(id=1, name="name4")

            result = conn.execute(i)
            eq_(result.inserted_primary_key, (1,))
            eq_(result.returned_defaults, None)

            eq_(
                conn.execute(
                    users.select().where(users.c.id == 10)
                ).fetchall(),
                [(10, "I'm a name")],
            ) 

def test_on_conflict_do_update_special_types_in_set(self):
        bind_targets = self.tables.bind_targets

        with testing.db.connect() as conn:
            i = insert(bind_targets)
            conn.execute(i, {"id": 1, "data": "initial data"})

            eq_(
                conn.scalar(sql.select([bind_targets.c.data])),
                "initial data processed",
            )

            i = insert(bind_targets)
            i = i.on_conflict_do_update(
                index_elements=[bind_targets.c.id],
                set_=dict(data="new updated data"),
            )
            conn.execute(i, {"id": 1, "data": "new inserted data"})

            eq_(
                conn.scalar(sql.select([bind_targets.c.data])),
                "new updated data processed",
            ) 

def test_reverse_eng_name(self):
        metadata = self.metadata
        engine = engines.testing_engine(options=dict(implicit_returning=False))
        for tname, cname in [
            ("tb1" * 30, "abc"),
            ("tb2", "abc" * 30),
            ("tb3" * 30, "abc" * 30),
            ("tb4", "abc"),
        ]:
            t = Table(
                tname[:57],
                metadata,
                Column(cname[:57], Integer, primary_key=True),
            )
            t.create(engine)
            with engine.begin() as conn:
                r = conn.execute(t.insert())
                eq_(r.inserted_primary_key, (1,)) 

def test_do_update_set_clause_none(self):
        i = insert(self.table_with_metadata).values(myid=1, name="foo")
        i = i.on_conflict_do_update(
            index_elements=["myid"],
            set_=OrderedDict([("name", "I'm a name"), ("description", None)]),
        )
        self.assert_compile(
            i,
            "INSERT INTO mytable (myid, name) VALUES "
            "(%(myid)s, %(name)s) ON CONFLICT (myid) "
            "DO UPDATE SET name = %(param_1)s, "
            "description = %(param_2)s",
            {
                "myid": 1,
                "name": "foo",
                "param_1": "I'm a name",
                "param_2": None,
            },
        ) 

def test_do_update_set_clause_literal(self):
        i = insert(self.table_with_metadata).values(myid=1, name="foo")
        i = i.on_conflict_do_update(
            index_elements=["myid"],
            set_=OrderedDict(
                [("name", "I'm a name"), ("description", null())]
            ),
        )
        self.assert_compile(
            i,
            "INSERT INTO mytable (myid, name) VALUES "
            "(%(myid)s, %(name)s) ON CONFLICT (myid) "
            "DO UPDATE SET name = %(param_1)s, "
            "description = NULL",
            {"myid": 1, "name": "foo", "param_1": "I'm a name"},
        ) 

def test_do_update_str_index_elements_target_one(self):
        i = insert(self.table_with_metadata).values(myid=1, name="foo")
        i = i.on_conflict_do_update(
            index_elements=["myid"],
            set_=OrderedDict(
                [
                    ("name", i.excluded.name),
                    ("description", i.excluded.description),
                ]
            ),
        )
        self.assert_compile(
            i,
            "INSERT INTO mytable (myid, name) VALUES "
            "(%(myid)s, %(name)s) ON CONFLICT (myid) "
            "DO UPDATE SET name = excluded.name, "
            "description = excluded.description",
        ) 

def test_do_update_index_elements_where_target(self):
        i = insert(self.table1, values=dict(name="foo"))
        i = i.on_conflict_do_update(
            index_elements=self.goofy_index.expressions,
            index_where=self.goofy_index.dialect_options["postgresql"][
                "where"
            ],
            set_=dict(name=i.excluded.name),
        )
        self.assert_compile(
            i,
            "INSERT INTO mytable (name) VALUES "
            "(%(name)s) ON CONFLICT (name) "
            "WHERE name > %(name_1)s "
            "DO UPDATE SET name = excluded.name",
        ) 

def test_do_update_add_whereclause(self):
        i = insert(self.table1, values=dict(name="foo"))
        i = i.on_conflict_do_update(
            constraint=self.excl_constr_anon,
            set_=dict(name=i.excluded.name),
            where=(
                (self.table1.c.name != "brah")
                & (self.table1.c.description != "brah")
            ),
        )
        self.assert_compile(
            i,
            "INSERT INTO mytable (name) VALUES "
            "(%(name)s) ON CONFLICT (name, description) "
            "WHERE description != %(description_1)s "
            "DO UPDATE SET name = excluded.name "
            "WHERE mytable.name != %(name_1)s "
            "AND mytable.description != %(description_2)s",
        ) 

def test_do_update_additional_colnames(self):
        i = insert(self.table1, values=dict(name="bar"))
        i = i.on_conflict_do_update(
            constraint=self.excl_constr_anon,
            set_=dict(name="somename", unknown="unknown"),
        )
        with expect_warnings(
            "Additional column names not matching any "
            "column keys in table 'mytable': 'unknown'"
        ):
            self.assert_compile(
                i,
                "INSERT INTO mytable (name) VALUES "
                "(%(name)s) ON CONFLICT (name, description) "
                "WHERE description != %(description_1)s "
                "DO UPDATE SET name = %(param_1)s, "
                "unknown = %(param_2)s",
                checkparams={
                    "name": "bar",
                    "description_1": "foo",
                    "param_1": "somename",
                    "param_2": "unknown",
                },
            ) 

def test_on_conflict_as_cte(self):
        i = insert(self.table1, values=dict(name="foo"))
        i = (
            i.on_conflict_do_update(
                constraint=self.excl_constr_anon,
                set_=dict(name=i.excluded.name),
                where=((self.table1.c.name != i.excluded.name)),
            )
            .returning(literal_column("1"))
            .cte("i_upsert")
        )

        stmt = select([i])

        self.assert_compile(
            stmt,
            "WITH i_upsert AS "
            "(INSERT INTO mytable (name) VALUES (%(name)s) "
            "ON CONFLICT (name, description) "
            "WHERE description != %(description_1)s "
            "DO UPDATE SET name = excluded.name "
            "WHERE mytable.name != excluded.name RETURNING 1) "
            "SELECT i_upsert.1 "
            "FROM i_upsert",
        ) 

def save_post_request(self):
        """Save the post request data into RDS."""
        try:
            insert_stmt = insert(StackAnalysisRequest).values(
                id=self.request_id,
                submitTime=self.submit_time,
                requestJson={'manifest': self.manifest},
                dep_snapshot=self.deps
            )
            do_update_stmt = insert_stmt.on_conflict_do_update(
                index_elements=['id'],
                set_=dict(dep_snapshot=self.deps)
            )
            rdb.session.execute(do_update_stmt)
            rdb.session.commit()
        except SQLAlchemyError as e:
            logger.exception("Error updating log for request {}, exception {}".format(
                self.request_id, e))
            raise RDBSaveException('Error while saving request {}'.format(self.request_id)) from e 

def set_update_state(self, entity_id: int, row: Any) -> None:
        t = self.UpdateState.__table__
        values = dict(pts=row.pts, qts=row.qts, date=row.date.timestamp(),
                      seq=row.seq, unread_count=row.unread_count)
        with self.engine.begin() as conn:
            conn.execute(insert(t)
                         .values(session_id=self.session_id, entity_id=entity_id, **values)
                         .on_conflict_do_update(constraint=t.primary_key, set_=values)) 

def cache_file(self, md5_digest: str, file_size: int,
                   instance: Union[InputDocument, InputPhoto]) -> None:
        if not isinstance(instance, (InputDocument, InputPhoto)):
            raise TypeError("Cannot cache {} instance".format(type(instance)))

        t = self.SentFile.__table__
        values = dict(id=instance.id, hash=instance.access_hash)
        with self.engine.begin() as conn:
            conn.execute(insert(t)
                         .values(session_id=self.session_id, md5_digest=md5_digest,
                                 type=_SentFileType.from_type(type(instance)).value,
                                 file_size=file_size, **values)
                         .on_conflict_do_update(constraint=t.primary_key, set_=values))