В python есть негласное правило — никогда не повторяйся. Чаще всего если в программе приходиться писать почти одно и то-же два раза, значит вы что-то сделали не так. Я приведу пример, как можно автоматизировать генерацию внешних API таким образом, что достаточно будет в одном месте в удобной и универсальной форме описать поддерживаемые вызовы, а все внешнее API для этих вызовов сделает написаный один раз код.

Итак мы пишем серверный компонент программы, который должен контролироваться внешними утилитами. Типичные варианты управления:

• CLI — административный интерфейс командной строки, так-же удобен для разработки
• REST — для других языков, WebUI & Co
• RCP в каком-то виде (thrift, PyRo, etc)

Нам нужна библиотека, которая позволит один раз задать интерфейсы API функций, сгенерирует по ним интерфейсы для всех внешних API, будет автоматически проверять входящие параметры и сделает удобочитаемую документацию. Для начала хватит.

Любая библиотека проектируется отталкиваясь от примеров ее использования.

Без подсветки синтаксиса

class Add(APICallBase):
    "Add two integers"
    class Params(object):
        params = Param([int], "list of integers to make a sum")

class Add(APICallBase): «Add two integers» class Params(object): params = Param([int], «list of integers to make a sum») def execute(self): return sum(self.params) class Sub(APICallBase): «Substitute two integers» class Params(object): params = Param((int, int), «substitute second int from first») def execute(self): return self.params[0] — self.params[1] class Ping(APICallBase): «Ping host» class Params(object): ip = Param(IPAddr, «ip addr to ping») num_pings = Param(int, «number of pings», default=3) def execute(self): res = subprocess.check_stdout(‘ping -c {0} {1}’.format(self.num_pings, self.ip)) return sum(map(float, re.findall(r’time=(d+.?d*)’, out))) / self.num_pings

Это желаемое описание API. Каждый API вызов наследует класс APICallBase, определяет внутренний класс Params, где экземплярами класса Param описывает параметры вызова и перегружает вызов execute, в котором выполняется вся работа. Этой информации более чем достаточно, что-бы сгенерировать все API и документацию пользуясь интроспекцией и генерацией объектов на лету.

Начнем с базы —
нужно уметь находить все классы, унаследованные от APICallBase. Это можно сделать через метаклассы

Показать код
Без подсветки синтаксиса

class APIMeta(type):

class APIMeta(type): # список всех API вызовов api_classes = [] def __new__(cls, name, bases, clsdict): new_cls = super(APIMeta, cls).__new__(cls, name, bases, clsdict) # пропускаем APICallBase if name != ‘APICallBase’: self.api_classes.append(new_cls) # all_params итерирует по всем параметрам этого вызова # передаем в параметры имена атрибутов, которым они присвоены # таким образом мы избегаем дублирования имени ‘ip’ в след строке # ip = Param(IPAddr, «ip addr to ping») # и других таких-же for name, param in new_cls.Params.__dict__.items(): param.name = name return new_cls # базовый класс для всех API вызовов class APICallBase(object): __metaclass__ = APIMeta def __init__(self, **dt): self._consume(dt) @classmethod def name(cls): return cls.__name__.lower() @classmethod def all_params(cls): return cls.Params.__dict__.values() def rest_url(self): return ‘/{0}’.format(self.name()) @classmethod def from_dict(cls, data): obj = cls.__new__(cls) obj._consume(data) return obj def _consume(self, data, from_strings=False): # этот метод заполняет экземпляр команды из словаря параметров # и проводит все необходимые проверки параметров required_param_names = set() all_param_names = set() for param in self.all_params(): if param.required(): required_param_names.add(param.name) all_param_names.add(param.name) # проверяем наличие лишних параметров data extra_params = set(data.keys()) — al
l_param_names if extra_params != set(): raise ValueError(«Extra parameters {0} for cmd {1}».format( ‘,’.join(extra_params), self.__class__.__name__)) # проверяем наличие в data всех необходимых параметров missed_params = required_param_names — set(data.keys()) if missed_params != set(): raise ValueError(«Missed parameters {0} for cmd {1}».format( ‘,’.join(missed_params), self.__class__.__name__)) # проверяем значение параметра или пребразовываем его из строки # (прошедшей из CLI) в целевой тип parsed_data = {} for param in self.all_params(): try: val = data[param.name] except KeyError: parsed_data[param.name] = param.default if from_strings: parsed_data[param.name] = param.from_cli(val) else: param.validate(val) parsed_data[param.name] = val # обновляем аттрибуты и возвращает объект self.__dict__.update(parsed_data) return self def to_dict(self): res = {} for param in self.all_params(): res[param.name] = getattr(self, param.name) return res def execute(self): # базовый метод для выполнения работы pass def __str__(self): res = «{0}({{0}})».format(self.__class__.__name__) params = [«{0}={1!r}».format(param.name, getattr(self, param.name)) for param in self.all_params()] return res.format(‘, ‘.join(params)) def __repr__(self): return str(self)

Классы для типов данных, используемых в Params

Без подсветки синтаксиса

# базовый класс для типов данных
class DataType(object):

< /span># базовый класс для типов данных class DataType(object): # проверить, про val принадлежит к денному типу def validate(self, val): return True # преобразовать val из формата для командной строки def from_cli(self, val): return None # параметры для парсера CLI def arg_parser_opts(self): return {} # список параметров определенного типа class ListType(DataType): def __init__(self, dtype): self.dtype = get_data_type(dtype) def validate(self, val): if not isinstance(val, (list, tuple)): return False for curr_item in val: if not self.dtype.valid(curr_item): return False return True def from_cli(self, val): return [self.dtype.from_cli(curr_item) for curr_item in val] def arg_parser_opts(self): opts = self.dtype.arg_parser_opts() opts[‘nargs’] = ‘*’ return opts # целое число class IntType(DataType): def validate(self, val): return isinstance(val, int) def from_cli(self, val): return int(val) def arg_parser_opts(self): return {‘type’: int}

Итак переходим к генерации API. Для начала — CLI

Без подсветки синтаксиса

def get_arg_parser():
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers()

def get_arg_parser(): parser = argparse.ArgumentParser() subparsers = parser.add_subparsers() for call in APIMeta.api_classes(): # для каждого вызова — свой вложенный парсер sub_parser = subparsers.add_parser(call.name(), help=call.__doc__) sub_parser.set_defaults(cmd_class=call) # проходим по всем параметрам и добавляем для них опции в CLI for param in call.all_params(): opts = {‘help’:param.help} # значение по умолчанию, если оно есть # _NoDef это специальный класс, что-бы отличать значение # None и полное отсутствие параметра if param.default is not _NoDef: opts[‘default’] = param.default opts.update(param.arg_parser_opts()) sub_parser.add_argument(‘—‘ + param.name.replace(‘_’, ‘-‘), **opts) return parser, subparsers

REST API с помощью CherryPy

Без подсветки синтаксиса

import cherrypy as cp
def get_cherrypy_server():

import cherrypy as cp def get_cherrypy_server(): class Server(object): pass # замыкание-обработчик для команды def call_me(cmd_class): # обмениваться данными будем через json @cp.tools.json_out() def do_call(self, opts): cmd = cmd_class.from_dict(json.loads(opts)) return cmd.execute() return do_call # добавляем к классу Server по методу для каждой команды # CherryPy будет их вызывать для обработки REST запросов for call in APIMeta.all_classes(APICallBase): setattr(Server, call.name(), cp.expose(call_me(call))) return Server

CherryPy довольно интересный веб-сервер, который использует интроспекцию и атрибуты классов для обработки HTTP запросов. Запрос вида http://localhost:8080/xyz?a=1&b=2 приведет к вызову Server.xyz(a=»1″, b=»2″), если такой есть и проброшен в web через cherrypy.expose.

Завершающий аккорд — функция main

Без подсветки синтаксиса

def main(argv=None):

def main(argv=None): # наполняем парсер CLI и разбираем командную строку argv = argv if argv is not None else sys.argv parser, subparsers = get_arg_parser() sub_parser = subparsers.add_parser(‘start-server’, help=»Start REST server») sub_parser.set_defaults(cmd_class=’start-server’) res = parser.parse_args(argv) cmd_cls = res.cmd_class # если пришел запрос на запуск сервера if cmd_cls == ‘start-server’: rest_server = get_cherrypy_server() cp.quickstart(rest_server()) else: # иначе конструируем объек-команду for opt in cmd_cls.all_params(): data = {} try: data[opt.name] = getattr(res, opt.name.replace(‘_’, ‘-‘)) except AttributeError: pass cmd = cmd_cls.from_dict(data) # если не определена переменная окружения REST_SERVER_URL rest_url = os.environ.get(‘REST_SERVER_URL’, None) if rest_url is None: # исполняем локально print «Local exec» print «Res =», cmd.execute() else: # иначе исполняем на сервере print «Remote exec» params = urllib.urlencode({‘opts’: json.dumps(cmd.to_dict())}) res = urllib2.urlopen(«http://{0}{1}?{2}».format(rest_url, cmd.rest_url(), params)).read() print «Res =», json.loads(res) return 0

Пробуем:

Без подсветки синтаксиса

$ python api.py -h
usage: api.py [-h] {add,sub,ping,start-server} ...