Bulk assignment of subscription adds all tiers rather than one (and expire isn't working)

Developer help
1

Hello,
I’m trying to bulk modify users to give them a Specific subscription tier with a specific expiration date.

But my script adds subscriptions to ALL tiers, and doesn’t set an expiration date. I’m failing to see my error. This topic seems related by as I’m on ghost pro, i can’t modify the core and am not sure it’s actually related.
Here’s my script so far:

import datetime
import jwt

# Configuration
ADMIN_URL = "https://SITE.ghost.io/ghost/api/admin"
API_KEY = "APIKEY" 
MEMBER_EMAIL = "test@gmail.com"
TIER_NAME = "Abonnement"
DURATION_DAYS = 30

# Generate JWT
def generate_jwt(api_key):
    key_id, secret = api_key.split(":")
    secret_bytes = bytes.fromhex(secret)

    now = datetime.datetime.now(datetime.UTC)  # ✅ Fix: Use timezone-aware datetime

    payload = {
        "iat": int(now.timestamp()),
        "exp": int((now + datetime.timedelta(minutes=5)).timestamp()),
        "aud": "/admin/"
    }

    token = jwt.encode(payload, secret_bytes, algorithm="HS256", headers={"kid": key_id})
    print(f"[DEBUG] Generated JWT Token: {token}")  # ✅ Print token for debugging
    return token


# Fetch Member by Email
def get_member_by_email(email):
    url = f"{ADMIN_URL}/members/?filter=email:{email}"
    headers = {"Authorization": f"Ghost {JWT_TOKEN}", "Content-Type": "application/json"}
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        members = response.json().get("members", [])
        return members[0] if members else None
    else:
        print(f"Error fetching member: {response.status_code}")
        return None

# Fetch Tier by Name
def get_tier_by_name(name):
    url = f"{ADMIN_URL}/tiers/"
    headers = {"Authorization": f"Ghost {JWT_TOKEN}", "Content-Type": "application/json"}
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        tiers = response.json().get("tiers", [])
        for tier in tiers:
            if tier["name"] == name:
                return tier
    print(f"Tier '{name}' not found.")
    return None

# Update Member with Complimentary Access
def update_member_with_comped_access(member, tier, duration_days):
    member_id = member["id"]
    tier_id = tier["id"]
    expires_at = (datetime.datetime.utcnow() + datetime.timedelta(days=duration_days)).isoformat() + 'Z'
    url = f"{ADMIN_URL}/members/{member_id}/"
    headers = {"Authorization": f"Ghost {JWT_TOKEN}", "Content-Type": "application/json"}
    data = {
        "members": [{
            "id": member_id,
            "comped": True,
            "expires_at": expires_at,
            "tiers": [{"id": tier_id}]
        }]
    }
    response = requests.put(url, headers=headers, json=data)
    if response.status_code == 200:
        print(f"Member {member['email']} updated with complimentary access to '{tier['name']}' until {expires_at}.")
    else:
        print(f"Error updating member: {response.status_code}, {response.text}")

# Main Execution
if __name__ == "__main__":
    JWT_TOKEN = generate_jwt(API_KEY)
    member = get_member_by_email(MEMBER_EMAIL)
    if member:
        tier = get_tier_by_name(TIER_NAME)
        if tier:
            update_member_with_comped_access(member, tier, DURATION_DAYS)
        else:
            print("Specified tier not found.")
    else:
        print("Member not found.")

Anyhelp would be appreciated

2

Ok, i solved my issue, my payload was wrong, here’s my script :

import requests
import datetime
import jwt
import json

# Configuration
ADMIN_URL = "https://mysite.ghost.io/ghost/api/admin"
API_KEY = "APIKEY" 
MEMBER_EMAIL = "test@gmail.com"
TIER_NAME = "Abonnement"
DURATION_DAYS = 30

# Generate JWT
def generate_jwt(api_key):
    key_id, secret = api_key.split(":")
    secret_bytes = bytes.fromhex(secret)
    now = datetime.datetime.now(datetime.UTC)  # Using recommended timezone-aware approach
    payload = {
        "iat": int(now.timestamp()),
        "exp": int((now + datetime.timedelta(minutes=15)).timestamp()),  # Increased to 15 minutes
        "aud": "/admin/"
    }
    token = jwt.encode(payload, secret_bytes, algorithm="HS256", headers={"kid": key_id})
    print(f"[DEBUG] Generated JWT Token: {token}")
    return token

# Fetch Member by Email
def get_member_by_email(email):
    url = f"{ADMIN_URL}/members/?filter=email:{email}"
    headers = {"Authorization": f"Ghost {JWT_TOKEN}", "Content-Type": "application/json"}
    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        members = response.json().get("members", [])
        if members:
            print(f"Found member: {members[0]['email']}")
            return members[0]
        else:
            print(f"No member found with email: {email}")
            return None
    else:
        print(f"Error fetching member: {response.status_code}, {response.text}")
        return None

# Fetch All Tiers
def get_all_tiers():
    url = f"{ADMIN_URL}/tiers/"
    headers = {"Authorization": f"Ghost {JWT_TOKEN}", "Content-Type": "application/json"}
    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        return response.json().get("tiers", [])
    else:
        print(f"Error fetching tiers: {response.status_code}, {response.text}")
        return []

# Get Tier by Name
def get_tier_by_name(name, all_tiers):
    for tier in all_tiers:
        if tier["name"] == name:
            print(f"Found tier: {tier['name']} (ID: {tier['id']})")
            return tier
    
    print(f"Tier '{name}' not found. Available tiers:")
    for tier in all_tiers:
        print(f" - {tier['name']} (ID: {tier['id']})")
    return None

# Update Member with Specific Tier Access ONLY
def update_member_with_exclusive_tier(member, tier, duration_days):
    member_id = member["id"]
    tier_id = tier["id"]
    
    # Set a specific expiration date (adding days to current UTC time)
    now = datetime.datetime.now(datetime.UTC)
    # Format using ISO 8601 format which Ghost expects
    expiry_date = now + datetime.timedelta(days=duration_days)
    expires_at = expiry_date.strftime("%Y-%m-%dT%H:%M:%S.000Z")
    print(f"[DEBUG] Setting expiration date to: {expires_at}")
    
    url = f"{ADMIN_URL}/members/{member_id}/"
    headers = {"Authorization": f"Ghost {JWT_TOKEN}", "Content-Type": "application/json"}
    
    # CRITICAL: Setting the member to have EXACTLY one tier 
    # Format exactly matching the Ghost Admin UI payload
    data = {
        "members": [{
            "id": member_id,
            "email": member["email"],
            "tiers": [{"id": tier_id, "expiry_at": expires_at}]  # ONLY this tier with expiry_at directly in the tier
        }]
    }
    
    print(f"[DEBUG] Updating member with data: {json.dumps(data, indent=2)}")
    
    response = requests.put(url, headers=headers, json=data)
    if response.status_code == 200:
        result = response.json()
        member_tiers = result["members"][0].get("tiers", [])
        tier_names = [t.get("name") for t in member_tiers]
        
        print(f"✅ Member {member['email']} updated with access until {expires_at}")
        print(f"Current tiers: {', '.join(tier_names)}")
        
        # Verify if we were successful
        if len(member_tiers) == 1 and member_tiers[0].get("id") == tier_id:
            print("✅ SUCCESS: Member has exactly the one tier we wanted!")
        else:
            print(f"⚠️ WARNING: Member still has {len(member_tiers)} tiers instead of just one.")
            print("You may need to contact Ghost support about this issue.")
    else:
        print(f"❌ Error updating member: {response.status_code}, {response.text}")

# Main Execution
if __name__ == "__main__":
    JWT_TOKEN = generate_jwt(API_KEY)
    
    # 1. Get the member
    member = get_member_by_email(MEMBER_EMAIL)
    if not member:
        print("❌ Cannot proceed: Member not found.")
        exit(1)
    
    # 2. Get all tiers first
    all_tiers = get_all_tiers()
    if not all_tiers:
        print("❌ Cannot proceed: Unable to fetch tiers.")
        exit(1)
    
    # 3. Find the specific tier we want
    tier = get_tier_by_name(TIER_NAME, all_tiers)
    if not tier:
        print("❌ Cannot proceed: Specified tier not found.")
        exit(1)
    
    # 4. Update the member with ONLY the specific tier
    update_member_with_exclusive_tier(member, tier, DURATION_DAYS)
1 Like