Skip to content

Commit

Permalink
new: use new Lacus endpoints to get stats
Browse files Browse the repository at this point in the history
  • Loading branch information
Rafiot committed Dec 7, 2023
1 parent e83d3b9 commit f817655
Show file tree
Hide file tree
Showing 2 changed files with 38 additions and 1 deletion.
2 changes: 1 addition & 1 deletion .github/workflows/mypy.yml
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ jobs:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.8', '3.9', '3.10', '3.11']
python-version: ['3.10', '3.11', '3.12']
name: Python ${{ matrix.python-version }} sample

steps:
Expand Down
37 changes: 37 additions & 0 deletions pylacus/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-

from base64 import b64decode
from datetime import datetime, date
from enum import IntEnum, unique
from importlib.metadata import version
from pathlib import Path
Expand Down Expand Up @@ -259,3 +260,39 @@ def get_capture(self, uuid: str, *, decode: bool=True) -> Union[CaptureResponse,
if not decode:
return response
return self._decode_response(response)

# # Stats and status of the lacus instance

def daily_stats(self, d: Optional[Union[str, date, datetime]]=None, /, *, cardinality_only: bool=True):
'''Get the stats for a specific day (only the last few days are stored in lacus).
:param cardinality_only: If True, only return the number of entries in each list (captures, retries, failed retries), instead of the URLs.
'''
if cardinality_only:
url_path = Path('daily_stats')
else:
url_path = Path('daily_stats_details')

if d:
if isinstance(d, (date, datetime)):
url_path /= d.isoformat()
else:
url_path /= d

r = self.session.get(urljoin(self.root_url, str(url_path)))
return r.json()

def db_status(self):
'''Gets the database status (number of keys, memory usage)'''
r = self.session.get(urljoin(self.root_url, 'db_status'))
return r.json()

def ongoing_captures(self, *, with_settings: bool=False):
r = self.session.get(urljoin(self.root_url, 'ongoing_captures'),
params={'with_settings': True} if with_settings else {})
return r.json()

def enqueued_captures(self, *, with_settings: bool=False):
r = self.session.get(urljoin(self.root_url, 'enqueued_captures'),
params={'with_settings': True} if with_settings else {})
return r.json()

0 comments on commit f817655

Please sign in to comment.