Skip to content

Commit f1d2eb2

Browse files
committed
Modified last call to mirror op-admin implementation
1 parent 96682d7 commit f1d2eb2

File tree

1 file changed

+17
-31
lines changed

1 file changed

+17
-31
lines changed

emission/pipeline/intake_stage.py

Lines changed: 17 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -215,7 +215,7 @@ def _get_and_store_range(user_id, trip_key):
215215
- user_id (str): The UUID of the user.
216216
- trip_key (str): The key representing the trip data in the time series.
217217
"""
218-
218+
time_format = 'YYYY-MM-DD HH:mm:ss'
219219
try:
220220
logging.info(f"Starting _get_and_store_range for user_id: {user_id}, trip_key: {trip_key}")
221221

@@ -245,43 +245,29 @@ def _get_and_store_range(user_id, trip_key):
245245

246246
logging.info(f"Total trips: {total_trips}, Labeled trips: {labeled_trips}")
247247
logging.info(type(user_id))
248-
# Retrieve last GET and PUT calls from stats/server_api_time
249-
docs_cursor = edb.get_timeseries_db().find({
250-
"metadata.key": "stats/server_api_time",
251-
"user_id" : user_id
252-
})
253248
logging.debug("Fetched API call statistics.")
254249

255-
last_get = None
256-
last_put = None
257-
258-
for doc in docs_cursor:
259-
api_call_name = doc.get("data", {}).get("name", "")
260-
api_call_ts = doc.get("data", {}).get("ts")
261-
262-
if not api_call_ts:
263-
logging.warning(f"Missing 'ts' in document: {doc}")
264-
continue
265-
266-
if api_call_name.startswith("GET_"):
267-
if not last_get or api_call_ts > last_get:
268-
last_get = api_call_ts
269-
logging.debug(f"Updated last_get to: {last_get}")
270-
elif api_call_name.startswith("PUT_"):
271-
if not last_put or api_call_ts > last_put:
272-
last_put = api_call_ts
273-
logging.debug(f"Updated last_put to: {last_put}")
274-
275-
# Determine the most recent call
276-
if last_get and last_put:
277-
last_call_ts = max(last_get, last_put)
278-
else:
279-
last_call_ts = last_get or last_put
250+
last_call_ts = ts.get_first_value_for_field(
251+
key='stats/server_api_time',
252+
field='data.ts',
253+
sort_order=pymongo.DESCENDING
254+
)
280255

281256
logging.info(f"Last call timestamp: {last_call_ts}")
282257

283258
# Update the user profile with pipeline_range, total_trips, labeled_trips, and last_call
284259
user = ecwu.User.fromUUID(user_id)
260+
if last_call_ts != -1:
261+
# Format the timestamp using arrow
262+
formatted_last_call = arrow.get(last_call_ts).format(time_format)
263+
# Assign using attribute access or the update method
264+
# Option 1: Attribute Assignment (if supported)
265+
# user.last_call = formatted_last_call
266+
267+
# Option 2: Using the update method
268+
user.update({
269+
"last_call": formatted_last_call
270+
})
285271
user.update({
286272
"pipeline_range": {
287273
"start_ts": start_ts,

0 commit comments

Comments
 (0)