Question: Use transaction ids to pull neuron ids /skids touched by users

[Mohinta2892]

Hi!

I have pulled the transaction history for all users interacting with a specific catmaid instance as below:

trans_df =  pymaid.get_transactions(remote_instance=rm_octo, range_length=500)

The df returned is as below:

transaction_id | execution_time | user_id | project_id | change_type | label  | user_anon
-- | -- | -- | -- | -- | -- | -- | --
2504056 | 2025-03-06 19:15:00 | 47 | 18 | Backend | annotations.add | Reviewer 6
2504054 | 2025-03-06 19:10:00 | 47 | 18 | Backend | annotations.add | Reviewer 6

Q: I now wish to use the transaction_id columns to figure out which neuron_ids/skids did these transactions modify or touch

I could not find any function that allows me to directly do this. How do I do answer Q?

Best,
Samia

[Mohinta2892]

I wrote the below solution based on available CATMAID APIs and taking inspiration from pymaid's base code:

import pymaid
import requests
import urllib

import pandas as pd
def get_transactions(range_start=None, range_length=25, remote_instance=None):
    """
    Retrieve transactions from Catmaid. Rewrite pymaid.get_transactions to include millisecond time."""
    remote_instance = pymaid.utils._eval_remote_instance(remote_instance)
    remote_transactions_url = remote_instance._get_transactions_url()

    desc = {'range_start': range_start, 'range_length': range_length}
    desc = {k: v for k, v in desc.items() if v is not None}
    remote_transactions_url += '?%s' % urllib.parse.urlencode(desc)

    data = remote_instance.fetch(remote_transactions_url)
    df = pd.DataFrame.from_dict(data['transactions'])

    user_list = pymaid.get_user_list(remote_instance=remote_instance)
    user_dict = user_list.set_index('id').login.to_dict()
    df['user'] = df.user_id.map(user_dict)

    # Preserve milliseconds and timezone
    df['execution_time'] = pd.to_datetime(df['execution_time'], format='%Y-%m-%dT%H:%M:%S.%f%z')

    return df


def _get_transactions_location_url(self, **GET):
    """Generate url to get transactions (GET)."""
    return self.make_url(self.project_id, 'transactions/location', **GET)

# Add the method to the CatmaidInstance class
setattr(pymaid.CatmaidInstance, '_get_transactions_location_url', _get_transactions_location_url)


def get_transactions_with_locations(remote_instance, project_id, range_length=500):
    """
    Retrieve all transactions (using pymaid.get_transactions) and add location details
    for each transaction.

    Args:
        remote_instance: The remote instance configuration for pymaid.
        project_id (int): The Catmaid project ID.
        range_length (int): How many transactions to retrieve.

    Returns:
        list: List of transaction dictionaries, each enriched with location data.
    """
    # Fetch transactions from pymaid.
    transactions = get_transactions(remote_instance=remote_instance, range_length=range_length)

    enriched_transactions = [] # list of dicts
    for i, transaction in transactions.iterrows(): # transaction is a df
        transaction_id = transaction.transaction_id
        execution_time = transaction.execution_time
        
        # print(transaction_id, execution_time)

        # Make sure the required keys are available.
        if not transaction_id or not execution_time:
            continue

        params = urllib.parse.urlencode({
        "transaction_id": transaction_id,
        "execution_time": execution_time
        })

        # Build URL for location details
        url = f"{remote_instance.server}/{remote_instance.project_id}/transactions/location?{params}"

        # If pymaid has a lower-level GET method, use it:
        try:
            # Use the built-in fetch method to retrieve data.
            location_data = remote_instance.fetch(url, return_type="json")

        except AttributeError:
            # Otherwise, fall back to using requests directly (ensure proper authentication as needed)
            location_response = requests.get(url, headers=pymaid.get_headers(remote_instance))
            location_response.raise_for_status()
            location_data = location_response.json()
        except Exception as e:
            print(f"Error fetching location data for transaction {transaction_id}: {e}")
            location_data = {'x': None,'y': None, 'z': None}

        # Append the location details to the transaction.
        transaction["location_data"] = location_data
        enriched_transactions.append(transaction)
    
    return enriched_transactions

Is it worth raising a PR to include it to pymaid ?

Best,
Samia

[schlegelp]

I take it you are after any kind of modification (nodes created, nodes modified, annotations added, etc)? In theory, you should be able to compile this from the various different endpoints (get_node_details, get_annotation_details, ...) but that wouldn't only show data that still exists - i.e. no node/annotation deletions. It's also possible that getting the entire transaction logs is ultimately faster?

In any event: neither Chris nor I have a stake in the game 😄 and if this function is useful to you, please feel free to PR!