====== Requests ====== ===== einfache http-Abfragen ===== * https://requests.readthedocs.io/en/master/user/quickstart/ * https://realpython.com/python-requests/ import requests #scriptname must not be requests.py or http.py! url="https://api.github.com" try: response = requests.get(url) # If the response was successful, no Exception will be raised response.raise_for_status() except requests.HTTPError as e: #print(f'HTTP error occurred: {e}') status_code = e.response.status_code print("Status:",status_code) except requests.Timeout as e: print('The request timed out') except requests.ConnectionError as e: print(f"connection-error {e}") except Exception as e: print(f'Other error occurred: {e}') # Python 3.6 else: print('Success!') #print(response.text) print(response.headers['content-type']) json_data=response.json() print(json_data['issues_url']) #get-param response = requests.get( 'https://api.github.com/search/repositories', params={'q': 'requests+language:python'}, headers={'Accept': 'application/vnd.github.v3.text-match+json'}, ) json_response = response.json() repository = json_response['items'][0] print(f'Repository name: {repository["name"]}') requests.post('https://httpbin.org/post', data={'key':'value'}) requests.put('https://httpbin.org/put', data={'key':'value'}) requests.delete('https://httpbin.org/delete') requests.head('https://httpbin.org/get') requests.patch('https://httpbin.org/patch', data={'key':'value'}) requests.options('https://httpbin.org/get') https://stackoverflow.com/a/47007419 print(response.status_code) Asynchrone (parallele) Abfragen: * https://medium.com/hackernoon/how-to-run-asynchronous-web-requests-in-parallel-with-python-3-5-without-aiohttp-264dc0f8546 * https://dev.to/matteo/async-request-with-python-1hpo ==== Authentifikation ==== response = requests.get(url, verify=False, auth=HTTPBasicAuth(self.user,self.pass)) ==== parallele Abfragen ==== import requests import asyncio import concurrent.futures class multiplerequests(): def basequery(self,url): print(url) response=requests.get(url) #print(response) ctype=response.headers['content-type'] #print(ctype) data=response.text #print(data) ret={"url":url,"type":ctype,"data":data} #print("basequery",ret) return ret #{"type":ctype,"data":data} async def multiquery_helper(self,urls): results=[] with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor: loop = asyncio.get_event_loop() futures = [ loop.run_in_executor( executor, self.basequery, u ) for u in urls ] for response in await asyncio.gather(*futures): #pass results.append(response) #print("mq_helper:",results) return results def multiquery(self,urls): loop = asyncio.get_event_loop() data=loop.run_until_complete(self.multiquery_helper(urls)) return data {{ :programming:python:apibase.py |}} ==== github ==== import requests token="your_token_here" headers = { 'User-Agent': 'Script', 'Authorization':'token '+token, } def ghrequest(url,session=None): # headers = { # 'User-Agent': 'Script', # 'Authorization':'token '+token, # } if not 'headers' in locals(): headers=None if session: response = session.get(url) else: response = requests.get(url, headers=headers) return response def getBranches(repo,session=None): url="https://api.github.com/repos/"+repo+"/branches?per_page=100" return ghrequest(url,session) def getForks(repo,session=None): url="https://api.github.com/repos/"+repo+"/forks" return ghrequest(url,session) def getClones(repo,session=None): url="https://api.github.com/repos/"+repo+"/traffic/clones" return ghrequest(url,session) session = requests.Session() session.headers.update(headers) repos={'frank-w/BPI-R2-4.14','frank-w/u-boot'} for r in repos: #res=getBranches(r) #print(res.status_code,"\n",res.headers,"\n",res.text) #res=getForks(r) #print(res.status_code,"\n",res.headers,"\n",res.text) res=getClones(r,session) print(res.status_code,"\n",res.headers,"\n",res.text)