Denna kod är utformad för att hämta artiklar från Finfos API-tjänst som kräver autentisering. Den hanterar autentisering, paginering och felhantering för att säkerställa att alla artiklar för en given leverantör kan hämtas och sparas i JSON-format. Koden är strukturerad i flera delar för att organisera funktionaliteten och göra den lättare att underhålla.
Koden saknar nödvändig struktur för backoff för både autentisering och artikelanropet, detta bör implementeras i kod som kör automatiserat.
Autentisering med OAuth2
För att hämta artiklar krävs en giltig access-token som erhålls genom OAuth2-autentisering. Koden innehåller funktioner för att både hämta en ny token och förnya token när den har löpt ut. Detta säkerställer att API-anrop alltid har en giltig token.
När koden körs, initieras en token-hämtning genom funktionen get_access_token(). Denna funktion gör ett HTTP POST-anrop till autentiseringsservern med client_secret, client_id, user och password. Svaret innehåller en access-token och information om när token löper ut. Denna information sparas och används för att autentisera efterföljande API-anrop.
Om en token har löpt ut, använder koden funktionen refresh_access_token() för att förnya token med hjälp av en refresh-token. Detta säkerställer kontinuerlig tillgång till API:t utan att behöva logga in manuellt igen.
Hämtning av Artiklar
Huvudfunktionen fetch_all_articles() ansvarar för att hämta alla artiklar för en specifik leverantör. Den hanterar paginering genom att successivt öka en offset och göra upprepade anrop tills alla sidor har hämtats. Funktionen kontrollerar även HTTP-svarskoder för att hantera eventuella fel som kan uppstå, till exempel om tillgång till en leverantörs artiklar nekas.
Artiklarna hämtas som en lista av JSON-objekt och varje artikel innehåller olika fält med information. Om API:et returnerar ett fel eller om ingen artikel hittas, hanteras detta genom att skriva ut lämpliga meddelanden.
Spara Artiklar till JSON-filer
När alla artiklar för en leverantör har hämtats, sparas de i en JSON-fil. Filnamnet innehåller leverantörens ID och den aktuella tidsstämpeln för att enkelt kunna identifiera datan. Detta gör det möjligt att effektivt spara och organisera hämtade artiklar för vidare bearbetning eller analys.
Du behöver ändra dessa fält så dom passar med dina uppgifter. Observera att körningen kommer att ta lång tid om du har många leverantörer.
-
client_secret = 'client secret'
-
user = 'your username'
-
password = 'your passwod'
import requests
import time
import json
from datetime import datetime, timedelta
# Configuration for OAuth2 authentication
access_token_url = 'https://sso.logiq.no/auth/realms/finfo/protocol/openid-connect/token'
client_id = 'finfo-api'
client_secret = 'client secret'
user = 'your username'
password = 'your passwod'
grant_type = 'password'
scope = 'openid'
# Token handling functions
def get_access_token():
"""Fetches a new access token from the authentication server."""
response = requests.post(access_token_url, auth=(client_id, client_secret), data={
'grant_type': grant_type,
'username': user,
'password': password,
'scope': scope
})
token_data = response.json()
token_data["expires_at"] = time.time() + token_data["expires_in"]
return token_data
def refresh_access_token(refresh_token):
"""Renews the access token using a refresh token."""
response = requests.post(access_token_url, auth=(client_id, client_secret), data={
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
'scope': scope
})
token_data = response.json()
token_data["expires_at"] = time.time() + token_data["expires_in"]
return token_data
def is_token_expired(token_data):
"""Checks if the current token has expired."""
return time.time() > token_data["expires_at"]
tokens = get_access_token()
def get_headers():
"""Ensures each API call uses a valid access token."""
global tokens
if is_token_expired(tokens):
print("Token has expired, refreshing...")
tokens = refresh_access_token(tokens['refresh_token'])
return {'Authorization': 'Bearer ' + tokens['access_token']}
def fetch_all_articles(supplier_id, from_date, to_date):
"""
Fetches all articles for a given supplier within a date range.
Handles pagination automatically by fetching 1000 articles at a time.
Args:
supplier_id: The Finfo supplier ID
from_date: Start date for article changes (YYYY-MM-DD)
to_date: End date for article changes (YYYY-MM-DD)
Returns:
List of articles or None if access is denied
"""
articles = []
offset = 0
limit = 1000
while True:
url = (f'https://api.finfo.se/api/v2.2/article'
f'?finfosupplierid={supplier_id}'
f'&changed-from-date={from_date}'
f'&changed-to-date={to_date}'
f'&offset={offset}'
f'&limit={limit}')
response = requests.get(url, headers=get_headers())
if response.status_code == 403:
print(f"Access denied for supplier {supplier_id}")
return None
if response.status_code != 200:
print(f"Error: Received status code {response.status_code} for supplier ID {supplier_id}")
break
try:
data = response.json()
except requests.exceptions.JSONDecodeError:
print(f"JSONDecodeError: Could not decode JSON for supplier ID {supplier_id}")
break
article_list = data.get('articleList', [])
if not article_list:
print(f"No articles found for supplier {supplier_id} in the specified date range.")
break
articles.extend(article_list)
if len(article_list) < limit:
break
offset += limit
return articles
# Main execution
def main():
"""
Main function that:
1. Gets today's date and yesterday's date
2. Fetches all suppliers for the authenticated user
3. For each supplier, fetches all articles modified in the last 24 hours
4. Saves the articles to JSON files named with supplier ID and timestamp
"""
# Get today's date and yesterday's date for the date range
to_date = datetime.now().strftime("%Y-%m-%d")
from_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
headers = get_headers()
response = requests.get('https://api.finfo.se/api/v1.0/suppliers/mysuppliers', headers=headers)
suppliers_data = response.json()
for supplier in suppliers_data:
finfoSupplierId = supplier['finfoSupplierId']
articles_data = fetch_all_articles(finfoSupplierId, from_date, to_date)
if articles_data:
current_time = datetime.now().strftime("%Y%m%d%H%M%S")
file_name = f'{finfoSupplierId}_{current_time}.json'
with open(file_name, 'w', encoding='utf-8') as jsonfile:
json.dump(articles_data, jsonfile, ensure_ascii=False, indent=4)
print(f"Articles for supplier {finfoSupplierId} have been saved to {file_name}.")
else:
continue
print("Articles have been saved to JSON files.")
if __name__ == "__main__":
main()