Skip to content

Instantly share code, notes, and snippets.

@staugur
Created September 14, 2016 08:15
Show Gist options
  • Save staugur/aeb4c5ffe9aebc0f7bae1d4eaea3d3ea to your computer and use it in GitHub Desktop.
Save staugur/aeb4c5ffe9aebc0f7bae1d4eaea3d3ea to your computer and use it in GitHub Desktop.
Based on the built-in function httplib requests similar to the python module similar to the URL module of the simple type of request
# coding:utf8
# taochengwei <[email protected]> @2016-09-13
__version__ = '0.1'
__doc__ = 'request url for json type'
import json, httplib, urllib, urlparse, logging
logging.basicConfig(
format='%(asctime)s %(levelname)s %(message)s',
datefmt='%a, %d %b %Y %H:%M:%S',
)
class RequestsException(httplib.HTTPException):
pass
class RequestsParamError(RequestsException):
pass
class Requests:
'''Request API URL type RESTful, return the dictionary data'''
def __init__(self, url=None, timeout=5, headers=None):
self.url = url
self.timeout = timeout
self.headers = headers or {"User-Agent": "SimpleRequestsAgent"}
self.jsonde = json.JSONDecoder()
self.jsonen = json.JSONEncoder()
def _url_split(self, url):
if isinstance(url, (str, unicode)):
_url = urlparse.urlparse(url)
return _url.scheme, _url.netloc, _url.path, _url.query
else:
raise RequestsParamError
def get(self, url=None, headers=None, timeout=None):
""" http method get function """
if headers and not isinstance(headers, dict):
raise RequestsParamError("headers not None or dict")
if timeout and not isinstance(timeout, int):
raise RequestsParamError("timeout not None or integer")
url = self.url or url
headers = headers or self.headers
timeout = timeout or self.timeout
scheme, netloc, path, query = self._url_split(url)
if query:
uri = '%s?%s' %(path, query)
else:
uri = path
netloc = netloc.split(':')
if len(netloc) == 2:
server, port = netloc
else:
server = netloc[0]
port = 443 if scheme == 'https' else 80
#print server,port,uri,headers
try:
if scheme == 'https':
httpClient = httplib.HTTPSConnection(host=server, port=port, timeout=timeout)
else:
httpClient = httplib.HTTPConnection(host=server, port=port, timeout=timeout)
httpClient.request(method='GET', url=uri, headers=headers)
response = httpClient.getresponse()
content = response.read() # here is str type
data = self.jsonde.decode(content) # convert str to dict
httpClient.close()
except Exception,e:
logging.error(e, exc_info=True)
data = {}
return data
def post(self, url=None, data=None, headers=None, timeout=None):
""" http method get function """
if headers and not isinstance(headers, dict):
raise RequestsParamError("headers not None or dict")
if timeout and not isinstance(timeout, int):
raise RequestsParamError("timeout not None or integer")
if data and not isinstance(data, dict):
raise RequestsParamError("data not None or dict")
url = self.url or url
headers = headers or self.headers
timeout = timeout or self.timeout
scheme, netloc, path, query = self._url_split(url)
if query:
uri = '%s?%s' %(path, query)
else:
uri = path
netloc = netloc.split(':')
if len(netloc) == 2:
server, port = netloc
else:
server = netloc[0]
port = 443 if scheme == 'https' else 80
#print server,port,uri,headers,data
try:
if scheme == 'https':
httpClient = httplib.HTTPSConnection(host=server, port=port, timeout=timeout)
else:
httpClient = httplib.HTTPConnection(host=server, port=port, timeout=timeout)
httpClient.request(method='POST', url=uri, body=data, headers=headers)
response = httpClient.getresponse()
content = response.read() # here is str type
data = self.jsonde.decode(content) # convert str to dict
httpClient.close()
except Exception,e:
logging.error(e, exc_info=True)
data = {}
return data
def put(self, url=None, data=None, headers=None, timeout=None):
""" http method get function """
if headers and not isinstance(headers, dict):
raise RequestsParamError("headers not None or dict")
if timeout and not isinstance(timeout, int):
raise RequestsParamError("timeout not None or integer")
if data and not isinstance(data, dict):
raise RequestsParamError("data not None or dict")
url = self.url or url
headers = headers or self.headers
timeout = timeout or self.timeout
scheme, netloc, path, query = self._url_split(url)
if query:
uri = '%s?%s' %(path, query)
else:
uri = path
netloc = netloc.split(':')
if len(netloc) == 2:
server, port = netloc
else:
server = netloc[0]
port = 443 if scheme == 'https' else 80
#print server,port,uri,headers,data
try:
if scheme == 'https':
httpClient = httplib.HTTPSConnection(host=server, port=port, timeout=timeout)
else:
httpClient = httplib.HTTPConnection(host=server, port=port, timeout=timeout)
httpClient.request(method='PUT', url=uri, body=data, headers=headers)
response = httpClient.getresponse()
content = response.read() # here is str type
data = self.jsonde.decode(content) # convert str to dict
httpClient.close()
except Exception,e:
logging.error(e, exc_info=True)
data = {}
return data
def delete(self, url=None, data=None, headers=None, timeout=None):
""" http method get function """
if headers and not isinstance(headers, dict):
raise RequestsParamError("headers not None or dict")
if timeout and not isinstance(timeout, int):
raise RequestsParamError("timeout not None or integer")
if data and not isinstance(data, dict):
raise RequestsParamError("data not None or dict")
url = self.url or url
headers = headers or self.headers
timeout = timeout or self.timeout
scheme, netloc, path, query = self._url_split(url)
if query:
uri = '%s?%s' %(path, query)
else:
uri = path
netloc = netloc.split(':')
if len(netloc) == 2:
server, port = netloc
else:
server = netloc[0]
port = 443 if scheme == 'https' else 80
#print server,port,uri,headers,data
try:
if scheme == 'https':
httpClient = httplib.HTTPSConnection(host=server, port=port, timeout=timeout)
else:
httpClient = httplib.HTTPConnection(host=server, port=port, timeout=timeout)
httpClient.request(method='DELETE', url=uri, body=data, headers=headers)
response = httpClient.getresponse()
content = response.read() # here is str type
data = self.jsonde.decode(content) # convert str to dict
httpClient.close()
except Exception,e:
logging.error(e, exc_info=True)
data = {}
return data
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment