Authentication and Data Retrieval Script Documentation

Overview

This Python script performs authentication and retrieves data from the Omnivoltaic GraphQL API. It consists of functions to authenticate a user, fetch a node ID based on a search term, and execute the main workflow.

Prerequisites

• Python 3.x
• requests library (pip install requests)

API Endpoints

• Authentication Endpoint: https://production-omnivoltaic-graphql-api.omnivoltaic.com/graphql
• Data Retrieval Endpoint: https://production-omnivoltaic-graphql-api.omnivoltaic.com/graphql

Functions

1. authenticate(email, password)

Authenticates a user and retrieves an access token.

Parameters:

• email (str): The user's email address
• password (str): The user's password

Returns: access_token (str) or None

Example:

import requests

def authenticate(email, password):
    url = "https://production-omnivoltaic-graphql-api.omnivoltaic.com/graphql"
    body = """
    mutation {
      signInUser(signInCredentials:{
        email: "%s"
        password: "%s"
      }){
        _id
        accessToken
      }
    }
    """ % (email, password)

    response = requests.post(url=url, json={"query": body})
    if response.status_code == 200:
        data = response.json()
        access_token = data.get('data', {}).get('signInUser', {}).get('accessToken')
        if access_token:
            return access_token
    return None

2. get_node_id(access_token, search_term)

Fetches a node ID based on a provided search term.

Parameters:

• access_token (str): The access token from authentication
• search_term (str): The search term to look up an item

Returns: node_id (str) or None

Example:

def get_node_id(access_token, search_term):
    query_customer = """
        query {
          getAllClientItems(
            search: "%s"
            first: 1
            queryorder: DESC
            assetaccount: false
          ) {
            page {
              edges {
                node {
                  _id
                  oemID
                  oemItemID
                  sellerItemID
                  description
                  actorName
                  createdAt
                  updatedAt
                }
              }
            }
          }
        }
    """ % search_term

    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.post(
        'https://production-omnivoltaic-graphql-api.omnivoltaic.com/graphql',
        json={'query': query_customer},
        headers=headers
    )
    if response.status_code == 200:
        data = response.json()
        edges = data.get('data', {}).get('getAllClientItems', {}).get('page', {}).get('edges', [])
        if edges:
            return edges[0]['node']['_id']
    return None

3. fetch_code_history(item_id, token)

Fetches the code history of an item from the GraphQL API.

Example:

import httpx
import asyncio

async def fetch_code_history(item_id: str, token: str):
    query = f"""
    query {{
        getItemsForDownloadById(ids: ["{item_id}"]) {{
            item {{
                _id
                oemID
                oemItemID
            }}
            codeHistory {{
                _id
                deleteStatus
                deleteAt
                codeType
                codeDays
                codeNumber
            }}
        }}
    }}
    """

    url = "https://production-omnivoltaic-graphql-api.omnivoltaic.com/graphql"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {token}",
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(url, json={"query": query}, headers=headers)

    if response.status_code != 200:
        raise Exception(f"HTTP error! Status: {response.status_code}")

    result = response.json()
    if "errors" in result:
        raise Exception(f"GraphQL errors: {result['errors']}")

    data = result.get("data", {}).get("getItemsForDownloadById", [])
    if not data:
        raise Exception(f"No data found for item ID: {item_id}")

    return data[0].get("codeHistory", [])

Security Considerations

• Do not hardcode credentials — Use environment variables or a configuration file
• Ensure HTTPS — Encrypt credentials in transit
• Do not expose access tokens in logs or error messages

Recommendations

• Implement environment variables for credentials
• Add exception handling for network failures
• Use GraphQL client libraries for better request handling
• Implement logging instead of print statements