add website stats and active users API endpoints

umami/umami/__init__.py

@@ -7,6 +7,8 @@
 from .impl import set_url_base, set_website_id, set_hostname  # noqa: F401, E402
 from .impl import verify_token_async, verify_token  # noqa: F401, E402
 from .impl import websites_async, websites  # noqa: F401, E402
+from .impl import website_stats, website_stats_async
+from .impl import active_users, active_users_async
 
 __author__ = 'Michael Kennedy <[email protected]>'
 __version__ = impl.__version__
@@ -18,6 +20,8 @@
     set_url_base, set_website_id, set_hostname,
     login_async, login, is_logged_in,
     websites_async, websites,
+    website_stats_async, website_stats,
+    active_users_async, active_users,
     new_event_async, new_event,
     new_page_view, new_page_view_async,
     verify_token_async, verify_token,

umami/umami/impl/__init__.py

@@ -10,6 +10,7 @@
 __version__ = '0.2.17'
 
 from umami.errors import ValidationError, OperationNotAllowedError
+from datetime import datetime
 
 url_base: Optional[str] = None
 auth_token: Optional[str] = None
@@ -92,7 +93,7 @@ async def login_async(username: str, password: str) -> models.LoginResponse:
         "password": password,
     }
     async with httpx.AsyncClient() as client:
-        resp = await client.post(url, data=api_data, headers=headers, follow_redirects=True)
+        resp = await client.post(url, json=api_data, headers=headers, follow_redirects=True)
         resp.raise_for_status()
 
     model = models.LoginResponse(**resp.json())
@@ -121,7 +122,7 @@ def login(username: str, password: str) -> models.LoginResponse:
         "username": username,
         "password": password,
     }
-    resp = httpx.post(url, data=api_data, headers=headers, follow_redirects=True)
+    resp = httpx.post(url, json=api_data, headers=headers, follow_redirects=True)
     resp.raise_for_status()
 
     model = models.LoginResponse(**resp.json())
@@ -540,6 +541,183 @@ def validate_login(email: str, password: str) -> None:
         raise ValidationError("Password cannot be empty")
 
 
+async def active_users_async(website_id: Optional[str] = None) -> int:
+    """
+    Retrieves the active users for a specific website.
+
+    Args:
+        website_id: OPTIONAL: The value of your website_id in Umami. (overrides set_website_id() value).
+
+
+    Returns: The number of active users.
+    """
+    validate_state(url=True, user=True)
+
+    website_id = website_id or default_website_id
+
+    url = f'{url_base}{urls.websites}/{website_id}/active'
+    headers = {
+        'User-Agent': user_agent,
+        'Authorization': f'Bearer {auth_token}',
+    }
+
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(url, headers=headers, follow_redirects=True)
+        resp.raise_for_status()
+
+    return int(resp.json().get("x", 0))
+
+
+def active_users(website_id: Optional[str] = None) -> int:
+    """
+    Retrieves the active users for a specific website.
+
+    Args:
+        website_id: OPTIONAL: The value of your website_id in Umami. (overrides set_website_id() value).
+
+
+    Returns: The number of active users.
+    """
+    validate_state(url=True, user=True)
+
+    website_id = website_id or default_website_id
+
+    url = f'{url_base}{urls.websites}/{website_id}/active'
+    headers = {
+        'User-Agent': user_agent,
+        'Authorization': f'Bearer {auth_token}',
+    }
+
+    resp = httpx.get(url, headers=headers, follow_redirects=True)
+    resp.raise_for_status()
+
+    return int(resp.json().get("x", 0))
+
+
+async def website_stats_async(start_at: datetime, end_at: datetime, website_id: Optional[str] = None, url: Optional[str] = None, 
+                              referrer: Optional[str] = None, title: Optional[str] = None, query: Optional[str] = None, event: Optional[str] = None, 
+                              host: Optional[str] = None, os: Optional[str] = None, browser: Optional[str] = None, device: Optional[str] = None, 
+                              country: Optional[str] = None, region: Optional[str] = None, city: Optional[str] = None) -> models.WebsiteStats:
+    """
+    Retrieves the statistics for a specific website.
+
+    Args:
+        start_at: Starting date as a datetime object.
+        end_at: End date as a datetime object.
+        website_id: OPTIONAL: The value of your website_id in Umami. (overrides set_website_id() value).
+        url: OPTIONAL: Name of URL.
+        referrer: OPTIONAL: Name of referrer.
+        title: OPTIONAL: Name of page title.
+        query: OPTIONAL: Name of query.
+        event: OPTIONAL: Name of event.
+        host: OPTIONAL: Name of hostname.
+        os: OPTIONAL: Name of operating system.
+        browser: OPTIONAL: Name of browser.
+        device: OPTIONAL: Name of device (ex. Mobile)
+        country: OPTIONAL: Name of country.
+        region: OPTIONAL: Name of region/state/province.
+        city: OPTIONAL: Name of city.
+
+    Returns: A WebsiteStatsResponse model containing the website statistics data.
+    """
+    validate_state(url=True, user=True)
+
+    website_id = website_id or default_website_id
+
+    api_url = f'{url_base}{urls.websites}/{website_id}/stats'
+
+    headers = {
+        'User-Agent': user_agent,
+        'Authorization': f'Bearer {auth_token}',
+    }
+    params = {
+        'start_at': int(start_at.timestamp() * 1000),
+        'end_at': int(end_at.timestamp() * 1000),
+    }
+    optional_params = {
+        'url': url,
+        'referrer': referrer,
+        'title': title,
+        'query': query,
+        'event': event,
+        'host': host,
+        'os': os,
+        'browser': browser,
+        'device': device,
+        'country': country,
+        'region': region,
+        'city': city,
+    }
+    params.update({k: v for k, v in optional_params.items() if v is not None})
+
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(api_url, headers=headers, params=params, follow_redirects=True)
+        resp.raise_for_status()
+
+    return models.WebsiteStats(**resp.json())
+
+
+def website_stats(start_at: datetime, end_at: datetime, website_id: Optional[str] = None, url: Optional[str] = None, 
+                  referrer: Optional[str] = None, title: Optional[str] = None, query: Optional[str] = None, event: Optional[str] = None, 
+                  host: Optional[str] = None, os: Optional[str] = None, browser: Optional[str] = None, device: Optional[str] = None, 
+                  country: Optional[str] = None, region: Optional[str] = None, city: Optional[str] = None) -> models.WebsiteStats:
+    """
+    Retrieves the statistics for a specific website.
+
+    Args:
+        start_at: Starting date as a datetime object.
+        end_at: End date as a datetime object.
+        url: OPTIONAL: Name of URL.
+        website_id: OPTIONAL: The value of your website_id in Umami. (overrides set_website_id() value).
+        referrer: OPTIONAL: Name of referrer.
+        title: (OPTIONAL: Name of page title.
+        query: OPTIONAL: Name of query.
+        event: OPTIONAL: Name of event.
+        host: OPTIONAL: Name of hostname.
+        os: OPTIONAL: Name of operating system.
+        browser: OPTIONAL: Name of browser.
+        device: OPTIONAL: Name of device (ex. Mobile)
+        country: OPTIONAL: Name of country.
+        region: OPTIONAL: Name of region/state/province.
+        city: OPTIONAL: Name of city.
+
+    Returns: A WebsiteStatsResponse model containing the website statistics data.
+    """
+    validate_state(url=True, user=True)
+
+    website_id = website_id or default_website_id
+
+    api_url = f'{url_base}{urls.websites}/{website_id}/stats'
+
+    headers = {
+        'User-Agent': user_agent,
+        'Authorization': f'Bearer {auth_token}',
+    }
+    params = {
+        'startAt': int(start_at.timestamp() * 1000),
+        'endAt': int(end_at.timestamp() * 1000),
+    }
+    optional_params = {
+        'url': url,
+        'referrer': referrer,
+        'title': title,
+        'query': query,
+        'event': event,
+        'host': host,
+        'os': os,
+        'browser': browser,
+        'device': device,
+        'country': country,
+        'region': region,
+        'city': city,
+    }
+    params.update({k: v for k, v in optional_params.items() if v is not None})
+
+    resp = httpx.get(api_url, headers=headers, params=params, follow_redirects=True)
+    resp.raise_for_status()
+
+    return models.WebsiteStats(**resp.json())
+
 def validate_state(url: bool = False, user: bool = False):
     """
     Internal helper function, not need to use this.

umami/umami/models/__init__.py

@@ -34,6 +34,17 @@ class Website(pydantic.BaseModel):
     teamId: typing.Optional[str] = None
     user: WebsiteUser
 
+class Metric(pydantic.BaseModel):
+    value: int
+    prev: int
+
+class WebsiteStats(pydantic.BaseModel):
+    pageviews: Metric
+    visitors: Metric
+    visits: Metric
+    bounces: Metric
+    totaltime: Metric
+
 
 class WebsitesResponse(pydantic.BaseModel):
     websites: list[Website] = pydantic.Field(alias='data')