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 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 @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 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 linn api = linnapi() use the linn api instance to make api requests, such as fetching data from linnworks example response data = await linn api make request("get", "api endpoint") this explains the usage of the linns api class, it is used to help interact with the linnworks api