AlmaAPIUpdate.py

# AlmaAPIUpdate.py Should be viewed as a template or reference implementation for how to structure an API-based sync between an SIS and Alma.
# It focuses on the core logic. In production, you would need to implement secure credential management, robust error handling, and potentially more complex field mapping and comparison logic.
# Should be viable and within api limits if user base is under 5000 records, but may require optimization or batching for larger institutions.

import requests
import json
import logging
from datetime import datetime
import time

# --- LOGGING SETUP ---
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("alma_sis_sync.log"),
        logging.StreamHandler()
    ]
)

# --- CONFIGURATION ---
# --- Should be in .env or secure vault in production ---
#ALMA_API_KEY = 'YOUR_API_KEY_HERE'  # Must have Read/Write permissions for 'Users'
ALMA_BASE_URL = 'https://api-na.hosted.exlibrisgroup.com/almaws/v1'
HEADERS = {
    'Authorization': f'apikey {ALMA_API_KEY}',
    'Accept': 'application/json',
    'Content-Type': 'application/json'
}

# Define which User Groups are managed by this sync. May need to be user type code
TARGET_USER_GROUPS = {'17, 18', '19'} 

# --- TEST / SAFETY FLAGS ---
TEST_MODE = True

# DRY_RUN=True means:
# - no POST
# - no PUT
# - no expirations
DRY_RUN = True

# Start with 10 users in sandbox
TEST_LIMIT = 10

# Prevent accidental expiration of real records
TEST_PRIMARY_ID_PREFIX = 'apitest_'

# Disable expiration workflow initially
DISABLE_ORPHAN_DEACTIVATION = True


# --- ALMA API CLIENT FUNCTIONS ---

def get_all_external_alma_users(limit_override=None, one_page_only=False):
    """
    Retrieves EXTERNAL users currently in Alma using limit/offset pagination.
    Returns dictionary structured as:
    { 'primary_id': {user_summary_data} }

    In TEST_MODE, we intentionally stop after first page.

    Note:
    The Alma /users list endpoint returns summary records. It does not reliably
    include full fields such as user_group. Fetch full user records before
    checking detailed fields.
    """

    logging.info("Retrieving existing external users from Alma...")

    all_users = {}

    limit = limit_override or 100
    offset = 0

    while True:

        url = (
            f"{ALMA_BASE_URL}/users"
            f"?source_system=EXTERNAL"
            f"&limit={limit}"
            f"&offset={offset}"
        )

        try:
            response = requests.get(url, headers=HEADERS)
            response.raise_for_status()

            data = response.json()

        except Exception as e:
            logging.error(
                f"Failed to retrieve user batch "
                f"starting at offset {offset}: {e}"
            )
            break

        users_batch = data.get('user', [])

        for user in users_batch:
            all_users[user['primary_id']] = user

        logging.info(
            f"Retrieved {len(all_users)} users so far..."
        )

        # TEST MODE:
        # stop after first request
        if one_page_only:
            break

        total_records = data.get('total_record_count', 0)

        if (
            total_records == ""
            or offset + limit >= int(total_records)
        ):
            break

        offset += limit

        # small rate-limit safety pause
        time.sleep(0.05)

    logging.info(
        f"Complete. Found {len(all_users)} "
        f"total external users in Alma."
    )

    return all_users


def get_full_user_record(primary_id):
    """
    Fetches complete user record.
    Necessary because Alma PUT calls require full payload.
    """

    url = f"{ALMA_BASE_URL}/users/{primary_id}"

    try:
        response = requests.get(url, headers=HEADERS)

        if response.status_code == 200:
            return response.json()

        logging.error(
            f"Failed to fetch full record "
            f"for user {primary_id}: {response.text}"
        )

        return None

    except Exception as e:
        logging.error(
            f"Exception retrieving user "
            f"{primary_id}: {e}"
        )

        return None


def create_user(user_payload):
    """
    Sends POST request to create Alma user.
    """

    primary_id = user_payload.get('primary_id')

    if DRY_RUN:
        logging.info(
            f"DRY RUN: would CREATE user {primary_id}"
        )
        return True

    url = f"{ALMA_BASE_URL}/users"

    try:
        response = requests.post(
            url,
            headers=HEADERS,
            json=user_payload
        )

        if response.status_code in [200, 201]:

            logging.info(
                f"Successfully created new user: "
                f"{primary_id}"
            )

            return True

        logging.error(
            f"Failed to create user "
            f"{primary_id}: {response.text}"
        )

        return False

    except Exception as e:
        logging.error(
            f"Exception creating user "
            f"{primary_id}: {e}"
        )

        return False


def update_user(primary_id, full_updated_payload):
    """
    Sends PUT request to update existing Alma user.
    """

    if DRY_RUN:
        logging.info(
            f"DRY RUN: would UPDATE user {primary_id}"
        )
        return True

    url = f"{ALMA_BASE_URL}/users/{primary_id}"

    try:
        response = requests.put(
            url,
            headers=HEADERS,
            json=full_updated_payload
        )

        if response.status_code == 200:

            logging.info(
                f"Successfully updated user: "
                f"{primary_id}"
            )

            return True

        logging.error(
            f"Failed to update user "
            f"{primary_id}: {response.text}"
        )

        return False

    except Exception as e:
        logging.error(
            f"Exception updating user "
            f"{primary_id}: {e}"
        )

        return False


def expire_and_deactivate_user(primary_id, full_record):
    """
    Expires and deactivates a full Alma user record.

    This function expects a full user record, not the summary record returned
    from the /users list endpoint.
    """

    if DISABLE_ORPHAN_DEACTIVATION:

        logging.info(
            f"Skipping expiration for {primary_id}; "
            f"DISABLE_ORPHAN_DEACTIVATION=True"
        )

        return True

    # only allow expirations for test accounts
    if not primary_id.startswith(TEST_PRIMARY_ID_PREFIX):

        logging.info(
            f"Skipping non-test user {primary_id}"
        )

        return True

    logging.info(
        f"Initiating expiration for orphaned user: "
        f"{primary_id}"
    )

    full_record['status'] = {
        'value': 'INACTIVE',
        'desc': 'Inactive'
    }

    full_record['expiry_date'] = (
        datetime.now().strftime('%Y-%m-%dZ')
    )

    return update_user(primary_id, full_record)


# --- TEST USER BUILDERS ---

def build_fake_new_user():
    """
    Creates simulated new SIS user payload.
    """

    return {
        "primary_id": f"{TEST_PRIMARY_ID_PREFIX}new001",
        "first_name": "API",
        "last_name": "TestNew",
        "account_type": {
            "value": "EXTERNAL"
        },
        "user_group": {
            "value": "17"
        },
        "expiry_date": "2029-06-30Z",
        "status": {
            "value": "ACTIVE"
        },
        "contact_info": {
            "email": [
                {
                    "preferred": "true",
                    "email_address": (
                        "apitest_new001@example.edu"
                    ),
                    "email_type": [
                        {
                            "value": "school"
                        }
                    ]
                }
            ]
        }
    }


def build_fake_update_user(alma_summary_user):
    """
    Creates simulated update payload.
    """

    return {
        "primary_id": alma_summary_user["primary_id"],
        "expiry_date": "2030-06-30Z"
    }


# --- SANDBOX TEST HARNESS ---

def run_sandbox_simulation():
    """
    Executes sandbox-safe test workflow.

    Workflow:
    1. Pull first 10 Alma users
    2. Simulate create
    3. Simulate update
    4. Simulate orphan expiration

    Important:
    Orphan group checks fetch full records because the /users list response
    does not reliably include user_group.
    """

    logging.info("====================================")
    logging.info("STARTING SANDBOX SIMULATION")
    logging.info("====================================")

    alma_users = get_all_external_alma_users(
        limit_override=TEST_LIMIT if TEST_MODE else None,
        one_page_only=TEST_MODE
    )

    sample_users = list(alma_users.values())

    if not sample_users:

        logging.error(
            "No Alma users retrieved from sandbox."
        )

        return

    logging.info(
        f"Using sample set of "
        f"{len(sample_users)} users."
    )

    # --- SIMULATE CREATE ---

    logging.info("------------------------------------")
    logging.info("SIMULATION: CREATE USER")
    logging.info("------------------------------------")

    fake_new_user = build_fake_new_user()

    create_user(fake_new_user)

    # --- SIMULATE UPDATE ---

    logging.info("------------------------------------")
    logging.info("SIMULATION: UPDATE USER")
    logging.info("------------------------------------")

    update_target = sample_users[0]

    fake_update = build_fake_update_user(update_target)

    target_primary_id = fake_update['primary_id']

    full_record = get_full_user_record(
        target_primary_id
    )

    if full_record:

        old_expiry = full_record.get(
            'expiry_date'
        )

        full_record['expiry_date'] = (
            fake_update['expiry_date']
        )

        logging.info(
            f"Would update expiry_date "
            f"from {old_expiry} "
            f"to {fake_update['expiry_date']}"
        )

        update_user(
            target_primary_id,
            full_record
        )

    # --- SIMULATE ORPHAN EXPIRATION ---

    logging.info("------------------------------------")
    logging.info("SIMULATION: ORPHAN EXPIRATION")
    logging.info("------------------------------------")

    # simulate empty SIS feed
    simulated_seen_ids = set()

    orphans = (
        set(alma_users.keys())
        - simulated_seen_ids
    )

    logging.info(
        f"Calculated "
        f"{len(orphans)} simulated orphan users."
    )

    orphan_counter = 0

    for orphan_id in orphans:

        full_record = get_full_user_record(orphan_id)

        if not full_record:

            logging.info(
                f"Skipping orphan {orphan_id}; "
                f"could not retrieve full record."
            )

            continue

        user_group_in_alma = (
            full_record
            .get('user_group', {})
            .get('value')
        )

        logging.info(
            f"User {orphan_id} has full-record "
            f"user_group value: {user_group_in_alma}"
        )

        if (
            user_group_in_alma
            not in TARGET_USER_GROUPS
        ):

            logging.info(
                f"Skipping orphan {orphan_id}. "
                f"Group '{user_group_in_alma}' "
                f"is outside targeted scope."
            )

            continue

        logging.info(
            f"SIMULATED orphan candidate: "
            f"{orphan_id}"
        )

        expire_and_deactivate_user(
            orphan_id,
            full_record
        )

        orphan_counter += 1

        time.sleep(0.05)

    logging.info("------------------------------------")
    logging.info("SIMULATION COMPLETE")
    logging.info("------------------------------------")

    logging.info(
        f"Users sampled: {len(sample_users)}"
    )

    logging.info(
        f"Orphans processed: {orphan_counter}"
    )


# --- CSV TO JSON HELPER ---

def convert_csv_to_json(
    input_csv,
    output_json
):
    """
    Converts flat CSV into Alma-shaped JSON payload.
    """

    import csv

    users = []

    with open(
        input_csv,
        newline='',
        encoding='utf-8'
    ) as f:

        reader = csv.DictReader(f)

        for row in reader:

            users.append({

                "primary_id": row["primary_id"],

                "first_name": row["first_name"],

                "last_name": row["last_name"],

                "account_type": {
                    "value": "EXTERNAL"
                },

                "user_group": {
                    "value": row["user_group"]
                },

                "expiry_date": row["expiry_date"],

                "status": {
                    "value": "ACTIVE"
                },

                "contact_info": {
                    "email": [
                        {
                            "preferred": "true",
                            "email_address": row["email"],
                            "email_type": [
                                {
                                    "value": "school"
                                }
                            ]
                        }
                    ]
                }
            })

    with open(
        output_json,
        'w',
        encoding='utf-8'
    ) as f:

        json.dump(
            users,
            f,
            indent=2
        )

    logging.info(
        f"Converted CSV to JSON: "
        f"{output_json}"
    )


# --- TEST EXECUTION HARNESS ---

if __name__ == "__main__":

    logging.info(
        "Sandbox Alma sync harness starting..."
    )

    # OPTIONAL:
    # convert_csv_to_json(
    #     "sis_feed.csv",
    #     "sis_feed.json"
    # )

    run_sandbox_simulation()