use creation time as reference for remove_old_requests

Author: alex75

cads_broker/database.py

@@ -505,27 +505,24 @@ def generate_adaptor_properties_hash(
     ).hexdigest()
 
 
-def get_adaptor_properties(
-    adaptor_properties_hash: str,
-    session: sa.orm.Session,
-) -> AdaptorProperties | None:
-    try:
-        statement = sa.select(AdaptorProperties.hash).where(
-            AdaptorProperties.hash == adaptor_properties_hash
-        )
-        return session.execute(statement).one()
-    except sqlalchemy.orm.exc.NoResultFound:
-        return None
-
-
-def add_adaptor_properties(
+def ensure_adaptor_properties(
     hash: str,
     config: dict[str, Any],
     form: dict[str, Any],
     session: sa.orm.Session,
-):
-    adaptor_properties = AdaptorProperties(hash=hash, config=config, form=form)
-    session.add(adaptor_properties)
+) -> None:
+    """Create adaptor properties (if not exists) or update its timestamp."""
+    try:
+        statement = (
+            AdaptorProperties.__table__.update()
+            .returning(AdaptorProperties.hash)
+            .where(AdaptorProperties.__table__.c.hash == hash)
+            .values(timestamp=datetime.datetime.now())
+        )
+        session.execute(statement).one()
+    except sqlalchemy.orm.exc.NoResultFound:
+        adaptor_properties = AdaptorProperties(hash=hash, config=config, form=form)
+        session.add(adaptor_properties)
 
 
 def add_event(
@@ -557,18 +554,12 @@ def create_request(
     request_uid: str | None = None,
 ) -> dict[str, Any]:
     """Create a request."""
-    if (
-        get_adaptor_properties(
-            adaptor_properties_hash=adaptor_properties_hash, session=session
-        )
-        is None
-    ):
-        add_adaptor_properties(
-            hash=adaptor_properties_hash,
-            config=adaptor_config,
-            form=adaptor_form,
-            session=session,
-        )
+    ensure_adaptor_properties(
+        hash=adaptor_properties_hash,
+        config=adaptor_config,
+        form=adaptor_form,
+        session=session,
+    )
     metadata["resources"] = resources
     metadata["qos_tags"] = qos_tags
     request = SystemRequest(

cads_broker/entry_points.py

@@ -26,57 +26,40 @@ def remove_old_requests(
         dbsettings = config.ensure_settings(config.dbsettings)
         connection_string = dbsettings.connection_string
     engine = sa.create_engine(connection_string)
+    time_delta = datetime.datetime.now() - datetime.timedelta(days=older_than_days)
     # clean system requests and (via cascading delete) events
     with engine.begin() as conn:
         database.logger.info("deleting old system_requests and events...")
         stmt = sa.delete(database.SystemRequest).where(
-            database.SystemRequest.finished_at
-            <= (sa.func.now() - datetime.timedelta(days=older_than_days))
+            database.SystemRequest.created_at <= time_delta
         )
         result = conn.execute(stmt)
         conn.commit()
         num_requests_deleted = result.rowcount
+        database.logger.info(
+            f"{num_requests_deleted} old system requests "
+            f"successfully removed from the broker database."
+        )
     # clean adaptor_properties
     with engine.begin() as conn:
         try:
-            database.logger.info(
-                "deleting old adaptor_properties (trying in a block)..."
-            )
+            database.logger.info("deleting old adaptor_properties...")
             stmt_ap_delete = sa.delete(database.AdaptorProperties).where(
-                database.AdaptorProperties.timestamp
-                <= (sa.func.now() - datetime.timedelta(days=older_than_days))
+                database.AdaptorProperties.timestamp <= time_delta
             )
             result = conn.execute(stmt_ap_delete)
             conn.commit()
             num_ap_deleted = result.rowcount
             database.logger.info(
-                f"{num_requests_deleted + num_ap_deleted} old records "
+                f"{num_ap_deleted} old adaptor properties "
                 f"successfully removed from the broker database."
             )
             return
         except sa.exc.IntegrityError:
-            # some requests still use some old adaptor_properties: do not return and continue
-            pass
-    database.logger.info("deleting old adaptor_properties...")
-    num_ap_deleted = 0
-    session_obj = sa.orm.sessionmaker(engine)
-    with session_obj.begin() as session:
-        stmt = sa.select(database.AdaptorProperties).where(
-            database.AdaptorProperties.timestamp
-            <= (sa.func.now() - datetime.timedelta(days=older_than_days))
-        )
-        for record in session.scalars(stmt):
-            try:
-                with session.begin_nested():
-                    session.delete(record)
-                    num_ap_deleted += 1
-            except sa.exc.IntegrityError:
-                pass
-    database.logger.info(
-        f"{num_requests_deleted + num_ap_deleted} old records "
-        f"successfully removed from the broker database."
-    )
-    return
+            database.logger.error(
+                "cannot remove some old records from table adaptor_properties."
+            )
+            raise
 
 
 @app.command()

tests/test_02_database.py

@@ -784,8 +784,33 @@ def test_create_request(session_obj: sa.orm.sessionmaker) -> None:
             db.SystemRequest.request_uid == request_dict["request_uid"]
         )
         request = session.scalars(statement).one()
+        adaptor_properties = request.adaptor_properties
+        assert adaptor_properties.hash == "adaptor_properties_hash"
+        initial_timestamp = adaptor_properties.timestamp
     assert request.request_uid == request_dict["request_uid"]
     assert request.user_uid == request_dict["user_uid"]
+    # create again a new request using the same adaptor properties: timestamp updated
+    with session_obj() as session:
+        request_dict = db.create_request(
+            user_uid="abc456",
+            setup_code="",
+            entry_point="sum",
+            request={},
+            metadata={},
+            process_id="submit-workflow",
+            session=session,
+            portal="c3s",
+            adaptor_config={"dummy_config": {"foo": "bar"}},
+            adaptor_form={},
+            adaptor_properties_hash="adaptor_properties_hash",
+        )
+        statement = sa.select(db.SystemRequest).where(
+            db.SystemRequest.request_uid == request_dict["request_uid"]
+        )
+        request = session.scalars(statement).one()
+        adaptor_properties = request.adaptor_properties
+        assert adaptor_properties.hash == "adaptor_properties_hash"
+        adaptor_properties.timestamp > initial_timestamp
 
 
 def test_get_request(session_obj: sa.orm.sessionmaker) -> None:

tests/test_90_entry_points.py

@@ -48,19 +48,12 @@ def mock_config(
     form: dict[str, Any] = {},
     timestamp: datetime.datetime | None = None,
 ):
-    if timestamp is None:
-        adaptor_properties = database.AdaptorProperties(
-            hash=hash,
-            config=config,
-            form=form,
-        )
-    else:
-        adaptor_properties = database.AdaptorProperties(
-            hash=hash,
-            config=config,
-            form=form,
-            timestamp=timestamp,
-        )
+    adaptor_properties = database.AdaptorProperties(
+        hash=hash,
+        config=config,
+        form=form,
+        timestamp=timestamp,
+    )
     return adaptor_properties
 
 
@@ -102,14 +95,14 @@ def test_init_db(postgresql: Connection[str], mocker) -> None:
 def prepare_db(
     session_obj,
     num_old_props_used_by_old,
-    num_old_props_used_by_recent,
     num_recent_props_used_by_recent,
     recent_days=3,
     old_days=370,
 ):
     now = sa.func.now()
     old_date = now - datetime.timedelta(days=old_days)
     recent_date = now - datetime.timedelta(days=recent_days)
+
     with session_obj() as session:
         # initialize
         session.query(database.SystemRequest).delete()
@@ -121,35 +114,25 @@ def prepare_db(
             adaptor_properties = mock_config(hash=old_hash, timestamp=old_date)
             session.add(adaptor_properties)
             request = mock_system_request(
-                adaptor_properties_hash=old_hash, finished_at=old_date
-            )
-            session.add(request)
-            event = database.Events(request_uid=request.request_uid)
-            session.add(event)
-        # add old properties with recent requests
-        for x in range(num_old_props_used_by_recent):
-            old_hash2 = f"old_hash_{x}_with_new_request"
-            adaptor_properties = mock_config(hash=old_hash2, timestamp=old_date)
-            session.add(adaptor_properties)
-            request = mock_system_request(
-                adaptor_properties_hash=old_hash2, finished_at=recent_date
+                adaptor_properties_hash=old_hash, created_at=old_date
             )
             session.add(request)
             event = database.Events(request_uid=request.request_uid)
             session.add(event)
+            session.commit()
         # add recent properties with recent requests
         for x in range(num_recent_props_used_by_recent):
             new_hash = f"new_hash_{x}"
             adaptor_properties = mock_config(hash=new_hash, timestamp=recent_date)
             session.add(adaptor_properties)
             request = mock_system_request(
-                adaptor_properties_hash=new_hash, finished_at=recent_date
+                adaptor_properties_hash=new_hash, created_at=recent_date
             )
             session.add(request)
             event = database.Events(request_uid=request.request_uid)
             session.add(event)
+            session.commit()
         # not existing case of recent properties with old requests
-        session.commit()
 
 
 def test_remove_old_records(session_obj: sa.orm.sessionmaker):
@@ -159,7 +142,6 @@ def test_remove_old_records(session_obj: sa.orm.sessionmaker):
     prepare_db(
         session_obj,
         num_old_props_used_by_old=0,
-        num_old_props_used_by_recent=6,
         num_recent_props_used_by_recent=5,
     )
     result = runner.invoke(
@@ -171,9 +153,9 @@ def test_remove_old_records(session_obj: sa.orm.sessionmaker):
         all_requests = session.query(database.SystemRequest).all()
         all_events = session.query(database.Events).all()
         all_props = session.query(database.AdaptorProperties).all()
-        assert len(all_requests) == 11
-        assert len(all_events) == 11
-        assert len(all_props) == 11
+        assert len(all_requests) == 5
+        assert len(all_events) == 5
+        assert len(all_props) == 5
 
     # test remove all (most recent is 3 days old)
     result = runner.invoke(
@@ -199,7 +181,6 @@ def test_remove_old_records(session_obj: sa.orm.sessionmaker):
     prepare_db(
         session_obj,
         num_old_props_used_by_old=10,
-        num_old_props_used_by_recent=0,
         num_recent_props_used_by_recent=5,
     )
     result = runner.invoke(
@@ -214,23 +195,3 @@ def test_remove_old_records(session_obj: sa.orm.sessionmaker):
         assert len(all_requests) == 5
         assert len(all_events) == 5
         assert len(all_props) == 5
-
-    # test remove only some requests (some old props have recent requests)
-    prepare_db(
-        session_obj,
-        num_old_props_used_by_old=10,
-        num_old_props_used_by_recent=3,
-        num_recent_props_used_by_recent=5,
-    )
-    result = runner.invoke(
-        entry_points.app,
-        ["remove-old-requests", "--connection-string", connection_string],
-    )
-    assert result.exit_code == 0
-    with session_obj() as session:
-        all_requests = session.query(database.SystemRequest).all()
-        all_events = session.query(database.Events).all()
-        all_props = session.query(database.AdaptorProperties).all()
-        assert len(all_requests) == 8
-        assert len(all_events) == 8
-        assert len(all_props) == 8