> ## Documentation Index
> Fetch the complete documentation index at: https://docs.lumos.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Upload Activity Records to Lumos

> Pull data from a third party system that collects user activity (e.g. IdP, CASB, Chrome Extension), and upload it into Lumos as last login or activity data.

<RequestExample>
  ```python Python theme={null}
  from datetime import datetime, timedelta
  import requests

  OKTA_TOKEN = "YOUR_OKTA_TOKEN"
  LUMOS_TOKEN = "YOUR_LUMOS_TOKEN"

  base_url = "https://YOUR_ENVIRONMENT.okta.com/api/v1"
  users_endpoint = "/users"
  logs_endpoint = (
      '/logs?filter=eventType+eq+"user.authentication.sso"'
      '+and+actor.id+eq+"{actor_id}"&since={since}&until={until}'
      '&sortOrder=DESCENDING&limit=1000'
  )

  def fetch_okta_users():
    users = []
    next_url = base_url + users_endpoint
    while next_url:
      response = requests.get(
      	next_url,
        headers={ "Authorization": f"SSWS {OKTA_TOKEN}" }
      )
      
      next_url = response.links.get("next", {}).get("url")
      users += response.json()
    return users

  def fetch_logs_for_user(user):
    logs = []
    checkpoint_start = datetime.utcnow() - timedelta(days=90)
    next_url = base_url + logs_endpoint.format(
      actor_id=user['id'],
      since=checkpoint_start.isoformat(),
      until=datetime.utcnow().isoformat(),
    )
      
    while next_url:
      response = requests.get(
      	next_url,
        headers={ "Authorization": f"SSWS {OKTA_TOKEN}" }
      )
      
      next_url = response.links.get("next", {}).get("url")
      logs += response.json()
    return logs

  def format_logs(logs):
    formatted_logs = []
    for log in logs:
    	timestamp = log.get("published")

      targets = log.get("target", [])
      app_identifiers = [
        target["id"]
        for target in targets
        if target.get("type") == "AppInstance"
      ]
      
      app_identifier = app_identifiers[0]
      actor_id = log.get("actor", {}).get("id", "")

      formatted_logs.append({
          "timestamp": timestamp,
          "event": { "type": "LOGIN" },
          "app": { "instance_identifier": app_identifier },
          "account": { "external_id": actor_id },
          "source_app_id": "OKTA_APP_ID"
      })
    return formatted_logs

  # Save activity records for each user
  for user in users:
    logs = fetch_logs_for_user(user)
    formatted_logs = format_logs(logs)
    
    # upload to Lumos
    requests.post(
      "https://api.lumos.com/activity_records",
      headers={ "Authorization": f"Bearer {LUMOS_TOKEN}" },
      json={
        "records": formatted_logs
      }
    )
  ```
</RequestExample>

<ResponseExample>
  ```json Response Example theme={null}
  {"success":true}
  ```
</ResponseExample>

<Steps>
  <Step title="Get a list of users">
    This will be used in order to upload activity for each user one at a time. This example fetches a list of Okta users to upload log data for.
  </Step>

  <Step title="Fetch activity records for each user">
    Perform an API call to get the list of authentication events from a third party. This example uses Okta SSO logs.
  </Step>

  <Step title="Format the logs">
    Lumos expects a timestamp, event type, and various identifiers that are used to match to the appropriate accounts in Lumos.
  </Step>

  <Step title="Publish activity records to Lumos">
    Lumos processes these as if they came from the source app (in this case, Okta). Processing usually happens immediately, but can take some time.
  </Step>
</Steps>