Pipelines
Connecting and Interacting with the Linnworks API

6min
This guide provides detailed instructions on how to connect to the Linnworks API, a cloud-based online inventory management system that allows users to manage online selling processes. The provided Python code establishes a connection, manages authentication, and interacts with Linnworks' data and services.
Code Overview
The Python code for connecting to the Linnworks API primarily consists of the LinnApi class and its methods. Here's an overview of its structure and functionality:
LinnApi Class
The LinnApi class is the base class for connecting to the Linnworks API. It performs three main tasks:
Initializes a session: This sets up a connection with the Linnworks API.
Sets up headers: The headers contain required information such as authorization tokens.
Handles token authorization: The class manages the process of obtaining and refreshing authorization tokens.
Acquiring a New Token
The get_new_token method is responsible for fetching and refreshing the authorization token required for API requests. It sends a POST request to the Linnworks API's authorization endpoint and updates the token when necessary. The token has a one-hour expiry time.
Python
def get_new_token(self) -> None:
   response = requests.post(self.token_url, params=self.refresh_data)
   if response.status_code != 200:
       raise Exception("Failed to get new token")
   response_data = response.json()
   new_access_token = response_data['Token']
   self.headers["Authorization"] = new_access_token
   self.token_expiry_time = datetime.now() + timedelta(hours=1)
def get_new_token(self) -> None:
   response = requests.post(self.token_url, params=self.refresh_data)
   if response.status_code != 200:
       raise Exception("Failed to get new token")
   response_data = response.json()
   new_access_token = response_data['Token']
   self.headers["Authorization"] = new_access_token
   self.token_expiry_time = datetime.now() + timedelta(hours=1)
﻿
Singleton Design Pattern
The get_instance function implements the Singleton design pattern, ensuring only one instance of the LinnApi class is created throughout the application’s lifecycle. This promotes efficient resource usage when making API requests.
Python
@staticmethod
def get_instance() -> 'LinnApi':
   if not hasattr(LinnApi, '_instance'):
       LinnApi._instance = LinnApi()
   return LinnApi._instance
@staticmethod
def get_instance() -> 'LinnApi':
   if not hasattr(LinnApi, '_instance'):
       LinnApi._instance = LinnApi()
   return LinnApi._instance
﻿
Asynchronous Requests
The make_request method is used to send asynchronous HTTP requests to the Linnworks API. It handles token refresh, retries, and various HTTP status codes. It returns response data in dictionary format. This function is reused ion the whole pipeline to make connections.
Python
async with session.request(method, full_url, headers=headers,
                           data=payload, params=params) as response:
   if response.status == 200:
       response.raise_for_status()
       response_text = await response.text()
       return json.loads(response_text)
   elif response.status == 401:
       main_log.error("Failed to make the request for %s. Status code: %s",
                      url, response.status)
       self.get_new_token()
       headers["Authorization"] = self.headers["Authorization"]
       await asyncio.sleep(10)
   elif response.status == 429:
       await asyncio.sleep(50)
   else:
       main_log.error("Failed to make the request for %s. Status code: %s",
                      url, response.status)
       text = await response.text()
       raise Exception(
           "Failed to make the request. Status code:", text)
async with session.request(method, full_url, headers=headers,
                           data=payload, params=params) as response:
   if response.status == 200:
       response.raise_for_status()
       response_text = await response.text()
       return json.loads(response_text)
   elif response.status == 401:
       main_log.error("Failed to make the request for %s. Status code: %s",
                      url, response.status)
       self.get_new_token()
       headers["Authorization"] = self.headers["Authorization"]
       await asyncio.sleep(10)
   elif response.status == 429:
       await asyncio.sleep(50)
   else:
       main_log.error("Failed to make the request for %s. Status code: %s",
                      url, response.status)
       text = await response.text()
       raise Exception(
           "Failed to make the request. Status code:", text)
﻿
Usage
Ensure that all Pre-requisites are in order and that you have the API credentials in your .env file.
Instantiate the LinnApi class:
Use the linn_api instance to make API requests, such as fetching data from Linnworks.
This explains the usage of the Linns API class, it is used to help Interact with the Linnworks Api.
Product Item Models
Inventory Items Retrieval Module