octopart.py (2880B)
1 #!/usr/bin/env python3 2 # -*- coding: utf-8 -*- 3 # vim:fenc=utf-8 4 5 import json 6 import sys 7 import urllib.error 8 import urllib.request 9 10 GET_PARTS_GRAPHQL_QUERY = """ 11 query MyPartSearch($q: String!, $filters: Map) { 12 search(q: $q, filters: $filters, limit: 5) { 13 total 14 results { 15 part { 16 id 17 mpn 18 aka_mpns 19 manufacturer { 20 name 21 } 22 short_description 23 octopart_url 24 specs { 25 attribute { 26 name 27 shortname 28 } 29 display_value 30 } 31 best_datasheet { 32 url 33 mime_type 34 } 35 best_image { 36 url 37 } 38 document_collections { 39 name 40 documents { 41 name 42 url 43 mime_type 44 } 45 } 46 sellers { 47 company { 48 name 49 } 50 } 51 } 52 } 53 } 54 } 55 """ 56 57 58 # copy pasted from: https://github.com/prisma-labs/python-graphql-client/blob/master/graphqlclient/client.py 59 class OctopartClient: 60 def __init__(self, token): 61 self.endpoint = "https://octopart.com/api/v4/endpoint" 62 if token is None: 63 print("I need OCTOPART_TOKEN") 64 sys.exit(1) 65 self.token = token 66 67 def execute(self, query, variables=None): 68 return self._send(query, variables) 69 70 def _send(self, query, variables): 71 data = {"query": query, "variables": variables} 72 headers = {"Accept": "application/json", "Content-Type": "application/json"} 73 74 if self.token is not None: 75 headers["token"] = "{}".format(self.token) 76 77 req = urllib.request.Request( 78 self.endpoint, json.dumps(data).encode("utf-8"), headers 79 ) 80 81 try: 82 response = urllib.request.urlopen(req) 83 return response.read().decode("utf-8") 84 except urllib.error.HTTPError as e: 85 print((e.read())) 86 raise e 87 88 def get_part(self, mpn): 89 query = GET_PARTS_GRAPHQL_QUERY 90 # parameters = { 91 # "q": "INA219AIDR", 92 # "filters": { 93 # "manufacturer_id": ["370"] 94 # } 95 # } 96 parameters = {"q": mpn} 97 98 resp = self.execute(query, parameters) 99 # print(json.dumps(json.loads(resp), indent = 2)) 100 return json.loads(resp)