Listening Session Tracking

Welcome back! In the previous lessons, you learned how to set up a backend for a music player app and how to serve track data through API endpoints. Now, let’s take the next step: tracking when users listen to tracks.

Tracking listening sessions is important for many reasons. It helps you understand user behavior, recommend new music, and even reward users for their activity. In real-world music apps, every time you play a song, the app records that event somewhere. In this lesson, you will learn how to log these listening sessions in your own app.

Before we dive into session tracking, let’s quickly remind ourselves of the current setup. You already have a Flask app that loads track data from a JSON file and serves it through API endpoints. Here’s a quick look at how the app and data are set up:

# app.py (snippet)
from flask import Flask, jsonify
from src.database import get_all_tracks, get_track_by_id

app = Flask(__name__)

@app.route('/api/tracks', methods=['GET'])
def get_tracks_endpoint():
    tracks_df = get_all_tracks()
    return jsonify(tracks_df.to_dict(orient='records'))

@app.route('/api/tracks/<track_id>', methods=['GET'])
def get_track_endpoint(track_id):
    track = get_track_by_id(track_id)
    return jsonify(track)

This setup allows you to retrieve all tracks or a single track by its ID. We will build on this foundation to add session tracking.

To log a listening session, you first need to make sure the track exists. This is done by accessing the track data. Here’s how you can get all tracks or a specific track by ID:

# src/database.py (snippet)
import pandas as pd
import os

TRACKS_FILE_PATH = os.path.join(os.path.dirname(__file__), '..', 'static', 'tracks.json')
_tracks_cache = None

def load_tracks_data():
    global _tracks_cache
    if _tracks_cache is not None:
        return _tracks_cache
    if not os.path.exists(TRACKS_FILE_PATH):
        _tracks_cache = pd.DataFrame()
        return _tracks_cache
    _tracks_cache = pd.read_json(TRACKS_FILE_PATH)
    if 'id' in _tracks_cache.columns:
        _tracks_cache['id'] = _tracks_cache['id'].astype(str)
    return _tracks_cache

def get_all_tracks():
    return load_tracks_data()

def get_track_by_id(track_id):
    tracks_df = get_all_tracks()
    if tracks_df.empty:
        return None
    track_match = tracks_df[tracks_df['id'] == str(track_id)]
    return track_match.iloc[0].to_dict() if not track_match.empty else None

Explanation:

• load_tracks_data() loads the track data from a JSON file and caches it for faster access.
• get_all_tracks() returns all tracks as a DataFrame.
• get_track_by_id(track_id) looks up a track by its ID and returns its details as a dictionary, or None if not found.

This is important because, before logging a session, you want to make sure the track actually exists.

Now, let’s see how to record a listening session. The goal is to save each session (user, track, timestamp, and source) in a CSV file. Here’s the function that does this:

# src/database.py (snippet)
import csv
from datetime import datetime

SESSIONS_FILE_PATH = os.path.join(os.path.dirname(__file__), '..', 'static', 'sessions.csv')

def log_listening_session(user_id, track_id):
    """Appends a listening session (user, track, timestamp) to the CSV file."""
    # Verify the requested track exists
    if get_track_by_id(track_id) is None:
        return False, f"Track {track_id} not found"

    fieldnames = ['user_id', 'track_id', 'timestamp', 'source']

    # Always rewrite header if file is missing or empty
    write_header = not os.path.exists(SESSIONS_FILE_PATH) or os.path.getsize(SESSIONS_FILE_PATH) == 0

    try:
        with open(SESSIONS_FILE_PATH, 'a', newline='') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            if write_header:
                writer.writeheader()

            writer.writerow({
                'user_id': user_id,
                'track_id': track_id,
                'timestamp': datetime.now().isoformat(),
                'source': 'api'
            })

        return True, "Session logged successfully"
    except Exception as e:
        print("Error logging session:", e)
        return False, "Failed to log session"

Explanation:

• The function starts by checking if the track exists using get_track_by_id.
• It defines the field names for the CSV file, which include: user_id, track_id, timestamp, and source.
• It checks if the CSV file is missing or empty — if so, it writes a header row.
• It appends a new row with the session data: the user who listened, the track that was played, the timestamp of the session, and the source (which is "api").
• It returns a success message if the write is successful, or an error if something goes wrong.

Example Output:
If you call log_listening_session("user123", "1") and the track exists, a new line like this will be added to sessions.csv:

user_id,track_id,timestamp
user123,1,2024-06-10T15:23:45.123456

This CSV file acts as a lightweight event log of all listening sessions.

To make this feature available to users, you need an API endpoint that receives session log requests and calls the logging function. Here’s how it’s done:

# app.py (snippet)
from src.database import log_listening_session

@app.route('/api/listen/<user_id>/<track_id>', methods=['POST'])
def log_session_endpoint(user_id, track_id):
    """API endpoint to log a listening session."""
    success, message = log_listening_session(user_id, track_id)
    
    if success:
        return jsonify({
            "message": message,
            "user_id": user_id,
            "track_id": track_id
        }), 201
    else:
        status_code = 404 if "not found" in message else 500
        return jsonify({"error": message}), status_code

Explanation:

• This endpoint listens for POST requests at /api/listen/<user_id>/<track_id>.
• It calls log_listening_session() with the provided user and track IDs.
• If the session is logged successfully, it returns a success message and a 201 status code.
• If there’s an error (like the track not being found), it returns an error message and the appropriate status code.

Example Output:
A successful request will return:

{
  "message": "Session logged successfully",
  "user_id": "user123",
  "track_id": "1"
}

If the track does not exist, you’ll get:

{
  "error": "Track 99 not found"
}

Let’s take a closer look at how your Flask route and logging function work together behind the scenes to create a listening session.

You now have two connected parts in your backend:

1. The database function: log_listening_session(user_id, track_id) This function lives in database.py and is responsible for:

• Validating the track exists (using get_track_by_id)
• Opening or creating sessions.csv
• Writing the session data (user, track, timestamp)
• Returning success or error status

It contains all the core business logic around session tracking. It makes no assumptions about how it’s called — which makes it reusable by other parts of your app (e.g., APIs or internal tools).

2. The Flask route: /api/listen/<user_id>/<track_id> This route lives in app.py. Its job is to:

• Receive a POST request from the outside world
• Extract the user_id and track_id from the URL
• Pass them to log_listening_session(...)
• Format the return message into a valid HTTP response (JSON + status code)

You can test your session logging route using curl in your terminal. Here's a sample POST request:

curl -X POST -s http://localhost:5001/api/listen/user123/track001

If the track exists in your tracks.json, this should return:

{
  "message": "Session logged successfully",
  "user_id": "user123",
  "track_id": "track001"
}

And your sessions.csv will now include a new row with the current timestamp:

user_id,track_id,timestamp
user123,track001,2024-06-28T12:34:56.789000

You can optionally improve the /api/listen/... route by adding input validation, such as:

• Checking if user_id is non-empty
• Restricting track_id to known formats
• Logging failed requests for debugging

Now that you're logging listening sessions, let’s take it one step further — what if you want to know how many times each track has been played?

This is common in real apps for generating “Top Charts” or personalized recommendations.

Here’s a function that counts how many times each track appears in the sessions.csv file:

# src/database.py (snippet)
def get_listen_counts():
    """
    Returns a dictionary mapping track_id to total listen count.
    """
    if not os.path.exists(SESSIONS_FILE_PATH):
        return {}
    
    try:
        sessions_df = pd.read_csv(SESSIONS_FILE_PATH)
        if 'track_id' not in sessions_df.columns:
            return {}
        
        counts = sessions_df['track_id'].value_counts().to_dict()
        return counts
    except Exception as e:
        print(f"Error reading session data: {e}")
        return {}

You can expose it with a new route like this:

# app.py (snippet)
from src.database import get_listen_counts

@app.route('/api/listens/counts', methods=['GET'])
def get_listen_counts_endpoint():
    """Returns a dictionary of track_id to total listen counts."""
    try:
        counts = get_listen_counts()
        return jsonify(counts)
    except Exception as e:
        # TODO: Handle unexpected errors gracefully
        print(f"Error in /api/listens/counts route: {e}")
        return jsonify({"error": "Internal server error"}), 500

A request to curl -s http://localhost:5001/api/listens/counts might return:

{
  "track001": 4,
  "track003": 2,
  "track005": 1
}

In this lesson, you learned how to track when a user listens to a track by:

• Checking if the track exists
• Logging the session to a CSV file with the user, track, and timestamp
• Creating an API endpoint to handle session logging requests

These are the basic building blocks for tracking user activity in a music app. In the next set of exercises, you’ll get hands-on practice with logging sessions and working with the API. This will help you reinforce what you’ve learned and prepare you for more advanced features in future lessons.