Get Zone Traffic Statistics

This example looks up traffic data in a selected Zone between two timestamps, and prints out the data in CSV format.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "requests",
# ]
# ///
"""Print CSV of zone traffic for a zone between two dates"""

import requests
import json
import urllib.parse

# Change to your Group and Zone names
group_name = 'Default'
zone_name = 'Default'

# Change to the ISO 8601 timestamps you want data between
start_time = '2026-02-13T16:00:00.00Z'
end_time = '2026-02-13T16:30:00.00Z'

# Enter your rWeb address here
BASE_URL = 'https://rweb.local.com'

HEADERS = {'Content-Type':'application/json',
        'Accept': 'application/json'}

# Enter your user credentials here
CREDENTIALS = {'login':'admin', 'password':'admin123'}

def url(resource, params=None):
    """Turn a resource into a url.

    Uses base_url and optionally encodes a hash of parameters"""
    querystring = ""
    if params != None:
        querystring = "?"+urllib.parse.urlencode(params)
    return BASE_URL + resource + querystring

def wrap(data):
    """Convert an object to an rWeb suitable json string

    Wraps the data in the standard rweb {'data': } object """
    return json.dumps({'data': data})

def login(session):
    """Log in to a session.

    The authentication cookie is saved to the session"""
    result = session.post(url('/authentication'),
            wrap(CREDENTIALS))
    assert(result.status_code == requests.codes.created)

def logout(session):
    """Close a session"""
    session.delete(url('/authentication'))

def find_group(session, name):
    """Find a group based on its name

    Query must result in only one group"""
    result = session.get(url('/groups', {'name': name}))
    assert(result.status_code == requests.codes.ok)

    group_list = json.loads(result.content)
    assert(group_list['rows'] == 1)

    return group_list['data'][0]

def find_zone(session, group, name):
    """Find a zone from a group object based on its name

    Query must result in only one zone"""
    result = session.get(url(group['zones_resource'], {'name': name}))

    zone_list = json.loads(result.content)

    print(f"Searching for zone: {name}")
    print(f"Found {zone_list['rows']} zone(s)")
    if zone_list['rows'] > 0:
        print(f"Zone names: {[z.get('name', 'N/A') for z in zone_list['data']]}")

    assert(zone_list['rows'] == 1), f"Expected 1 zone, found {zone_list['rows']}"
    return zone_list['data'][0]

def main():
    with requests.session() as session:
        session.headers = HEADERS
        # This is insecure but may be needed if you haven't set up an SSL certificate for rWeb yet
        session.verify = False
        login(session)

        group = find_group(session, group_name)
        zone = find_zone(session, group, zone_name)

        traffic_types = ['all', 'ipv4', 'ipv6', 'tcp', 'udp', 'icmp', 'other',
                'tcp_syn', 'tcp_syn_ack', 'tcp_ack', 'tcp_fin', 'tcp_rst',
                'ip_frag']
        columns = ['packets_in', 'packets_out', 'bits_in', 'bits_out']

        data = []

        # Query zone for each traffic type we care about
        for traffic_type in traffic_types:
            result = session.get(url(zone['traffic_resource'], {
                'start_time': start_time,
                'end_time': end_time,
                'traffic_type': traffic_type}))

            #Combine dates by traffic type
            for i, row in enumerate(json.loads(result.content)['data']):
                if len(data) <= i:
                    data.append({})
                data[i][traffic_type] = row

        # Print Header line
        print('timestamp,', end=' ')
        print(', '.join(['{0}_{1}'.format(traffic_type, column)
            for traffic_type in traffic_types
            for column in columns
            ]))

        # Print data by traffic type
        for i in data:
            timestamp = i['all']['timestamp_text']
            print(timestamp+',', end=' ')
            print(', '.join([str(i[traffic_type][column])
                for traffic_type in traffic_types
                for column in columns
                ]))

        logout(session)

if __name__ == '__main__':
    main()