Skip to content

Instantly share code, notes, and snippets.

@nkoneko
Created May 29, 2019 18:13
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save nkoneko/3329cebd75f1c6152c9def09676e1f91 to your computer and use it in GitHub Desktop.
Save nkoneko/3329cebd75f1c6152c9def09676e1f91 to your computer and use it in GitHub Desktop.
import urllib.parse
import urllib.request
import base64
import json
class APIMeta(type):
def __init__(cls, name, bases, nmspc):
super(APIMeta, cls).__init__(name, bases, nmspc)
@property
def URL(cls):
return f'{cls.__BASE_URL__}{cls.__PATH__}'
class APIClientMeta(APIMeta):
def __init__(cls, name, bases, nmspc):
super(APIClientMeta, cls).__init__(name, bases, nmspc)
if not hasattr(cls, '_apis'):
cls._apis = dict()
n = name.lower()
if n.endswith('api'):
n = n[:-3]
cls._apis[n] = cls
for base in bases:
bn = base.__name__.lower()
if bn in cls._apis:
del cls._apis[bn]
class APIBase(object):
def _find_path_params(self, url):
i = url.find('{')
if i > -1:
j = url[i:].find('}')
if j > -1:
l = self._find_path_params(url[i+j+1:])
name = url[i+1:i+j]
l.append(name)
return l
else:
return []
def _req(self, **params):
if 'content_type' in params:
self.content_type = params['content_type']
del params['content_type']
if 'method' in params:
method = params['method']
if method not in self.__class__.__METHODS__:
raise RuntimeError()
del params['method']
elif len(self.__class__.__METHODS__) == 1:
method = self.__class__.__METHODS__[0]
else:
raise RuntimeError()
url = self.__class__.URL
path_params = self._find_path_params(url)
for path_param in path_params:
if path_param not in params:
raise RuntimeError()
url = url.replace('{' + path_param + '}', str(params[path_param]))
del params[path_param]
if method == 'GET':
qstring = urllib.parse.urlencode(params)
if qstring:
url = f'{url}?{qstring}'
req = urllib.request.Request(url, method=method, headers=self.headers)
else:
req = urllib.request.Request(url, data=json.dumps(params), method=method, headers=self.headers)
return req
class BasicAuthAPIBase(APIBase, metaclass=APIMeta):
def __init__(self, user, password):
self.user = user
self.password = password
self.content_type = 'application/json'
@property
def headers(self):
encoded = base64.b64encode(f'{self.user}:{self.password}'.encode('ascii')).decode('ascii')
return {
'Authorization': f'Basic {encoded}',
'Accept': 'application/json',
'Content-Type': self.content_type
}
def __call__(self, **params):
req = self._req(**params)
with urllib.request.urlopen(req) as res:
return json.loads(res.read())
class PlainAPIBase(APIBase, metaclass=APIMeta):
def __init__(self):
self.content_type = 'application/json'
@property
def headers(self):
return {
'Accept': 'application/json',
'Content-Type': self.content_type
}
def __call__(self, **params):
req = self._req(**params)
with urllib.request.urlopen(req) as res:
return json.loads(res.read())
class BasicAuthAPIClientBase(object):
def __init__(self, user, password):
self.user = user
self.password = password
def __getattr__(self, attr):
if attr in self.__class__._apis:
return self.__class__._apis[attr](self.user, self.password)
else:
return self.__dict__[attr]
class PlainAPIClientBase(object):
def __getattr__(self, attr):
if attr in self.__class__._apis:
return self.__class__._apis[attr]()
else:
return self.__dict__[attr]
# 実装するのはここから。
class ExampleAPIBase(BasicAuthAPIBase):
__BASE_URL__ = 'https://example.com'
class ExampleAPIClient(BasicAuthAPIClientBase, metaclass=APIClientMeta):
pass
class FooAPI(ExampleAPIBase, ExampleAPIClient):
__PATH__ = '/foo'
__METHODS__ = ['GET']
class BarAPI(ExampleAPIBase, ExampleAPIClient):
__PATH__ = '/bar/{id}/item/{itemid}'
__METHODS__ = ['GET']
class Example2APIBase(PlainAPIBase):
__BASE_URL__ = 'https://example2.com'
class Example2APIClient(PlainAPIClientBase, metaclass=APIClientMeta):
pass
class HogeAPI(Example2APIBase, Example2APIClient):
__PATH__ = '/hoge'
__METHODS__ = ['GET', 'POST']
class HugaAPI(Example2APIBase, Example2APIClient):
__PATH__ = '/huga/{id}'
__METHODS__ = ['GET', 'POST', 'PUT', 'DELETE']
client = ExampleAPIClient('user', 'password')
print(client.bar(id=1, itemid=2, foo='bar'))
print(client.foo())
client2 = Example2APIClient()
print(client2.hoge(method='GET'))
print(client2.huga(method='POST', id=21, name='Ogawa', email='nekomura@example.com'))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment