Реализация паттерна Service Locator на Python

Вряд ли стоит описывать важность знания паттернов проектирования в современной разработке  приложений и сервисов. Но вот все ли паттерны мы знаем? На слуху - разумеется синглтон, декоратор, адаптер, фасад, прокси, фабрика и прочие. А вот IoC-контейнер (инверсия контроля) и DI (внедрение зависимости) - многие ли слышали, многие ли применяли? Между тем, не смотря на споры вокруг ценности и применимости IoC, он вполне имеет право на существование и при правильном применении может существенно облегчить жизнь разработчику.

В данной статье я хочу рассмотреть простой пример реализации паттерна Service Locator (частного случая IoC) на замечательном языке Python. Этот паттерн представляет из себя дополнительный абстрактный слой, реализованный в виде контейнера, хранящего “таблицу переходов” и функционал для её управления: добавление, удаление и подмена параметров перехода.

Применение Service Locator’а помогает избегать развесистых цепочек условных операторов и усложнения алгоритмов: достаточно сохранить в контейнере класс и потом, по мере надобности, подменять его на другие классы, при том что интерфейс обращения к нему остаётся одним и тем же - по заранее заданному ключу (тут я назвал его алиасом).

Собственно: сам класс, реализующий паттерн Service Locator:

# coding: utf-8

"""Service Locator: a simple python implementation."""

from importlib import import_module


class ServiceLocatorError(Exception): pass


class ServiceLocator(object):

    u"""Класс для работы с контейнером названий объектов.

    Представляет собой реализацию паттерна Service Locator.
    """

    # словарь для хранения сервисных классов
    _objects = {}

    @classmethod
    def set_class(cls, alias, class_path):
        u"""Добавить новый сервисный класс в контейнер.

        :param alias: алиас сервисного класса
        :type alias: str
        :param class_path: полное имя класса
        :type object_name: str
        """
        assert isinstance(alias, basestring), type(alias)
        assert isinstance(class_path, basestring), type(class_path)

        cls._objects[alias] = class_path

    @classmethod
    def get_class(cls, alias, default=None):
        u"""Найти имя класса в сервис-локаторе по его алиасу.

        :param alias: Алиас класса
        :type alias: str
        :return: полное имя объекта или None
        """
        assert isinstance(alias, basestring), type(alias)

        return cls._objects.get(alias, default)

    @classmethod
    def fabricate(cls, alias, *args, **kwargs):
        u"""Импортировать и создать экземпляр класса по его алиасу.

        :param alias: Алиас сервисного класса
        :type alias: str
        :param args: аргументы для конструктора класса
        :param kwargs: именованные аргументы для конструктора класса
        :return: инстанс класса
        """
        assert isinstance(alias, basestring), type(alias)

        full_class_name = cls._objects.get(alias)
        if full_class_name:
            full_path, class_name = full_class_name.rsplit('.', 1)

            try:
                module = import_module(full_path)
                return getattr(module, class_name)(*args, **kwargs)
            except (ImportError, AttributeError) as err:
                raise ServiceLocatorError(err)

        raise ServiceLocatorError(u"Unknown alias: %s" % alias)

    @classmethod
    def remove(cls, alias):
        u"""Удаление алиаса из Service Locator’а.

        :param alias: алиас класса
        :type alias: str
        """
        assert isinstance(alias, basestring), type(alias)

        if cls._objects.get(alias):
            del cls._objects[alias]

Из приведённого кода видно, что о каждом классе хранится следующая информация:

    - алиас - короткое название или ещё какой условный синоним,

    - полное имя - имя сервисного класса и полный путь к нему, по которому он будет вызываться (пакеты, подпакеты и модуль),  например: “foo_package.bar_module.ProviderClazz”.

Пример использования:

...
# сохранять(добавлять в контейнер) можно так:
ServiceLocator.set_class(
    'test_provider',
    'legacy.provider.Provider'
)

# а можно даже так:
if some_flag:
    ServiceLocator.set_class(
        'test_provider',
        'foo.provider.FooProvider'
    )
else:
    ServiceLocator.set_class(
        'test_provider',
        'bar.provider.BarProvider'
    )

        ...

# получение экземпляра класса
provider = ServiceLocator.fabricate(
    'test_provider',
     a=10, b=20, c=30
)
        
print provider.func()

Реализация Service Locator на Java - здесь 

Перехватчики для Git на Python - это просто.

Не так давно решал интересную задачу: нужно было разработать простенький hook-скрипт для git. А если конкретнее, то отслеживать макс.длину измененных строк в файлах, добавленных в текущий коммит. Для этих целей служит специальный хук pre-commit (подробнее о них - тут). Разбираться с gitpy не было времени и желания. К тому же захотелось написать самому. Оказалось, это не так сложно. Практически все примеры хуков на Python, которые я нашёл в инете - все используют модуль subprocess для работы с git. И вообще, работа с git через Python (и не только) сводится к запуску команд git и парсингу результатов. Решением стал пакет simplegit, содержащий специальный класс Git():

...
class Git(object):
    """ class for a GIT interface
    """
    def _call_git(self, *params):
        """ call a git command with params
        """
        p = subprocess.Popen(
            ["git"] + list(params),
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=False
            )
        exit_code = p.wait()
        if exit_code != 0:
            raise GitException("Error: git exit code = %s" % exit_code)

        return p.stdout.read().strip()

    def config(self, *options):
        """ run: git config 
        """
        return self._call_git("config", *options)

    def get_param(self, param, default=None):
        """ get one param from git config
        """
        try:
            res = self.config("--get", param)
            return res
        except GitException:
            return default

    def set_param(self, param, value, filename=""):
        """ set one param
        """
        if filename:
            self.config("-f", filename, param, value)
        else:
            self.config(param, value)

    def del_param(self, param):
        """ remove parameter
        """
        self.config("--unset", param)

    def check_git(self):
        """ check: git is installed?
        """
        try:
            self._call_git("--version")
            return True
        except OSError:
            return False

    def get_files(self):
        """ return a file list... or empty list
        """
        ret = self._call_git(
             'diff',
             '--cached',
             '--diff-filter=AM',
             '-U0',
             '--name-only')
        if ret:
            return ret.split('\n')
        else:
            return []
...

Чтобы написать скрипт, реализующий перехватчик (hook) - необходимо создать свой класс, унаследовав его от simplegit.Git и дописав необходимый функционал:

class GitHook(Git):
    """ simple git hook for a client side
    """
    def check_max_length(self, filename, default_line_length=80):
        """ check max length of source lines
        """
        # get a config
        max_line_length = self.get_param("pre-commit.max-line-length",
                                         default_line_length)
        
        try:
            max_line_length = int(max_line_length)
        except:
            max_line_length = default_line_length

        # get a lines and check
        for n,s in self.get_diff_rows(filename):
            length = len(s)
            if length > max_line_length:
                print "%s:%d:Line has length %d but %d is allowed" % 
                    (filename, n, length, max_line_length)
                print "%s[...]" % s[:80]
                return False

        return True

    def check_file(self, filename):
        if self.get_param("pre-commit.max-line-length.enabled") != "false":
            res = self.check_max_length(filename)
            if not res:
                return False
            
        # ... another check-routine place here...
        return True

    def check(self):
        """ check all files in a commit
        """
        for filename in self.get_files():
            print filename
            if not self.check_file(filename):
                return False

        # ... otherwise check routines place here...
        return True


if __name__ == "__main__":
    print "[PRE-COMMIT HOOK] Check a files...."
    hook = GitHook()
    if hook.check():
        exit(0)
    else:
        exit(1)

Примечание: проект тестировался на Ubuntu 12.04 и FreeBSD 9.1, на версии Python 2.7.3
Исходники проекта с инсталляторами и юнит-тестами на github: тут

Мемоизация в Python: пробуем на зуб.

Как-то мне посчастливилось натолкнуться на хабрапост, посвященный мемоизации. На тот момент это слово было мне ещё не знакомым и я даже слегка растерялся.
Однако уже после первых строк я сразу понял, что скрывается под этим мудрёным словом - это же старое, доброе, классическое ("тёплое ламповое" ;-) ) табулирование функций, но подведённое под паттерн проектирования и с более удобным (в контексте языка Python) интерфейсом.

Почитав ещё несколько постов на эту же тему я увидел в представленных реализациях "фатальный недостаток" тут же кинулся переписывать самостоятельно, на примере расчёта факториала:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from functools import wraps
import cPickle
import math
import time

mem = {}

def save():
    f = open('memoization.dat', 'wb')
    cPickle.dump(mem, f)
    f.close()

def load():
    global mem
    try:
        f = open("memoization.dat", 'rb')
        mem = cPickle.load(f)
        f.close()
    except:
        pass

def make_key(name, *args, **kw):
    # создание строкового ключа на основе имени функции и её входных параметров
    key = name + '|'
    key += "|".join([str(p) for p in args]) + "|"
    key += "|".join(["%s:%s" % (str(k),str(kw[k])) for k in sorted(kw.keys())])
    return key

def memoization(func):
    # специальный декоратор для кеширования
    @wraps(func)
    def wrap(*args, **kwargs):
        # формируем ключ
        key = make_key(func.__name__, *args,**kwargs)
        # поиск в кеше
        if key not in mem:
            mem[key] = func(*args,**kwargs)
        return mem[key]

    return wrap

def tracetime(func):
    # специальный декоратор для оценки времени работы функций
    @wraps(func)
    def wrap(*args, **kwargs):
        t1 = time.time()
        res = func(*args, **kwargs)
        t2 = time.time()
        print func.__name__, '/ dt=', t2-t1
        return res

    return wrap

@memoization
def fact(x):
    # мемоизированный факториал
    return math.factorial(x)

# простой тест: сумма длин строковых представлений факториалов
# из заданного диапазона

@tracetime
def test1(a,b):    
    L = 0
    for x in range(a,b):
        L += len(str(fact(x)))

    return L

@tracetime
def test2(a,b):
    L = 0
    for x in range(a,b):
        L += len(str(math.factorial(x)))

    return L

a = 2000
b = 3000

# первый вызов - данных в словаре нет
print test1(a,b)
# обычный расчёт
print test2(a,b)
# при повторном вызове - мемоизация будет уже задействована
print test1(a,b)

Однако отличия от увиденного ранее - минимальны. Такова уж специфика задачи и особенности Python (да, досадно вышло - хочется изобрести свой велосипед, а даже код получается почти такой же, как у всех!). Отличие от прочих вариантов - словарь описан снаружи и пригоден к сериализации с дальнейшим сохранением в файл посредством pickle.
Описанный выше подход - красив и лаконичен, но слишком обобщён и идеален, годится лишь как демонстрация красоты Python и наглядности подхода.
Теперь добавим несколько смачных ложек дёгтя.

1) В приведенном примере для хранения значений функций используется простой словарь, который хранится в памяти. Его размер определяется ограничениями используемой операционной системы. То есть запуская скрипт с мемоизацией на разных конфигурациях железа и ОС мы не можем расчитывать, что нам всегда будет хватать памяти на одних и тех же объёмах вычислений.

2) Может так получиться, что расчёты будут производиться внутри параллельно выполняющихся процессов - в этом случае, приведенный метод потребует доработки: словарь mem придётся сделать разделяемой переменной, что может несколько усложнить приложение.

3) Области определения функций и действительно ли надо хранить так много значений?
Например, простые числа или числа Фибоначчи. Куда удобнее хранить их не в словаре, а в упорядоченном списке - и использовать двоичный поиск для проверки наличия значений. Из этого следует очевидный вывод, что способ хранения и доступа определяется особенностями функции.

Всё-таки, если бы вдруг встала острая необходимость мемоизации - действительно важная функция (или их некоторое множество) считалась долго, нужна была часто и вероятность повторного вызова с таким же набором входных параметров была бы заметно отлична от нуля - пожалуй я выбрал бы для этого какую-нибудь базу данных, MySQL или PostgreSQL - не суть важно. Храни сколько хочешь, доставай хоть из треда, хоть из процесса. А обращение также обернул бы в декоратор.

Python для параллельных вычислений

XXI век давно уже наступил и параллельные вычисления перестали быть чем-то сложным и недостижимым. Многие платформы и языки имеют средства для реализации распараллеливания и замечательный язык Python не является исключением.
Python предоставляет разнообразные инструменты для реализации многозадачности: многопоточную, многопроцессную и т.н. сопрограммы. Не хотелось бы вдаваться в глубокую теорию и описывать отличия и особенности многопоточной и многопроцессной многозадачностей - всё это есть в книгах, которые без труда находятся в гугле или яндексе. Вкратце стоит лишь отметить, что для математических вычислений лучше всего подходит именно многопроцессная многозадачность, которая реализована в Python благодаря модулю multiprocessing.

Перейдём к практике. Допустим есть такая нехитрая задача:

Найти минимальное натуральное число с суммой цифр 80, которое делится на 1237.

(Это несколько упрощённый аналог задачи с замечательного сайта Диофант)

Наверняка, такую задачу можно решить аналитически, обладая хорошим математическим образованием. Но нас интересует, как решить её с помощью компьютера и Python.
Если применить простой последовательный перебор в одном цикле, решаться она будет долго. Имея многоядерный процессор такую задачу можно легко распараллелить и решать тоже перебором, но параллельно ;).
Поскольку предугадать на каком интервале находится искомое число трудно - поступим следующим образом: создадим функцию поиска, принимающую на входе интервал чисел и выдающую результат. Будем запускать параллельно несколько процессов с этой функцией с последовательными и относительно небольшими интервалами и проверять, найдено ли решение.

Итак, исходный код (использовался Python 2.7.2):

#!/usr/bin/env python/
# -*- coding: utf-8 -*-

import multiprocessing
import time

# очередь для сбора результатов
result_queue = multiprocessing.Queue()

# блокировка
output_lock = multiprocessing.Lock()

def sumdig(v):
    """ Возвращает сумму цифр числа v.
    """
    Sum = 0
    remain = v
    while remain > 10:
        Sum += remain % 10
        remain = remain // 10
    Sum += remain
    return Sum

def worker(num, n1,n2, qres):
    """
    Поиск числа с суммой цифр =80 в заданном диапазоне [n1, n2].
   
    num - номер воркера. Сугубо для того, чтобы отличать их друг от друга
    n1, n2 - диапазон значений
    qres - объект Queue, очередь для накопления результатов
    """
    t1 = time.time()
    x = n1
    res = 0
    while x <= n2:
        sm = sumdig(x)
        if sm == 80:
            res = x
            break
        x += 1237
    t2 = time.time()   
   
    # запросим блокировку для вывода на экран результатов работы воркера
    output_lock.acquire()
   
    print 'worker #%d Res=%d  time: %0.4f ' % (num, res, float(t2-t1)),
    print ' n1=%d, n2=%d' % (n1,n2)

    # отпустим блокировку
    output_lock.release()
   
    if res > 0:
        qres.put_nowait(res)


if __name__ == '__main__':

    # число воркеров принимаем равным числу ядер
    worker_count = multiprocessing.cpu_count()

    start_value = 0
    M = 100000
    while True:
        jobs = []
        for i in xrange(worker_count):
            p = multiprocessing.Process(target=worker,
                                        args=(i, start_value + 1237*M*i + 1237,
                                                 start_value + 1237*M*(i+1),
                                                 result_queue))
            jobs.append(p)
            p.start()

        # ожидаем завершения работы всех воркеров
        for w in jobs:
            w.join()

        # начальное значение для следующей итерации
        start_value = start_value + 1237*M*(worker_count)
        print "start_value=", start_value


        if result_queue.empty():
            print "next iteration..."
        else:
            print "--- THE END ---"
            break

    # из очереди выберем все ответы и найдём самый минимальный
    res = []
    while not result_queue.empty():
        res.append(result_queue.get())

    print 'RESULT=', min(res) 

Некоторые замечания по исходному коду:

1) Каким выбрать число воркеров - заранее предсказать сложно. При малом его количестве - задача будет решаться медленнее. При большом - можно здорово подгрузить свою систему, что повлечет замедление работы не только воркеров, но и прочих процессов - так, например, при тестировании этого примера на 6 и более воркерах (четырёхядерный процессор i3, Windows7 и было ещё запущено много всякого типа Aptana Studio, Bat! и 4 окна Firefox c кучей вкладок) я наблюдал такие эффекты - с заметными задержками открывались программы и перетаскивались окна. Диспетчер задач показывал почти 100% загрузку всех ядер процессора.
    Практика показала, что макс. число воркеров должно выбираться эмпирически, с учётом загрузки процессора и среднего времени работы воркеров (в нашем случае, интервалы поиска чисел не следует задавать слишком длинными) и относительно оптимальным значением явлется число равное числу ядер процессора.

2) Для нормального отображения хода работы, а именно чтобы сообщения не накладывались друг на друга, искажая смысл - нужно блокировать вывод на консоль. Пожалуй, тема блокировок стоит того, чтобы рассмотреть её отдельно.

3) В данном примере, из соображений упрощения, для хранения воркеров задействован обычный список (jobs). Хотя на самом деле в модуле multiprocessing есть реализация пула процессов, обладающая широким набором функций: multiprocessing.Pool().

Как видно из примера - многозадачность, основанная на механизме многопроцессности, не такая уж сложная и страшная штука, а Python вполне себе годен для решения математических задач путём распараллеливания.

Работаем с оракловыми курсорами в Python с помощью библиотеки cx_Oracle

Курсорная переменная может возвращаться в качестве результата хранимой процедуры, написанной на PL/SQL. Например, вот такой:

create or replace procedure test_cur(
  a number,
  b number,
  p_ret out sys_refcursor)
is
begin
  open p_ret for select a + level-1 v from dual
          connect by level <= b;
end;

Для работы с СУБД Oracle из Python есть весьма хорошая библиотека cx_Oracle . С курсорами работа построена там очень просто - они объявляются как обычные типизированные переменные библиотеки, в данном случае с типом курсора. Всё очень просто, смотрим на примере:

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import cx_Oracle as db

try:
    conn = db.connect('user', 'password', 'server')
except  db.DatabaseError, err:
    print 'DB Error: ', err
    exit()

cur = conn.cursor()

# так объявляется курсорная переменная
ret = cur.var(db.CURSOR)

# Не забываем, что есть dir(), с чьей помощью можно узнать очень
# много полезного об инстансе курсорной переменной
print 'ret: ', dir(ret)
print 'ret.getvalue(): ', dir(ret.getvalue())
   
cur.execute('''begin test_cur(1, 20, :ret); end; ''', ret=ret)

# описание полей запроса
print ret.getvalue().description

# А вот по этому можно уже пройтись через for ;-)
print ret.getvalue().fetchall()

# не забываем закрывать за собой соединение с Ораклом
conn.close()

Кстати, вот тут лежит неплохой мануал по cx_Oracle на русском языке.

Ещё один способ "заточки пилы"

Не все из нас работают в стартапах или научных проектах, в творческих коллективах и по SCRUM-методикам. У многих программистов профессиональная деятельность связана с довольно скучными вещами типа сопровождения отчётности и всякой бухгалтерии-логистики.
В этом нет ничего противоестественного или фатального, но иногда могут появляться ощущения однообразия и затягивания в "профессиональную яму".

Для лечения таких симптомов есть масса интересных способов. Ищущий, да найдёт! Недавно я открыл для себя еще один, увлекательный и полезный. Нет, это не онлайн-шахматы или подкидной дурак на мэйл.ру. Это ресурс с несколькими сотнями логических, математических, а также физико-химических задачек.

А открыл я для себя Diofant.ru

Это совместный проект компаний "Интернет-Университет Информационных Технологий (INTUIT.ru)" и издательства "Открытые Системы".
Чтобы решать задачи, регистрироваться не обязательно. Но чтобы узнать правильность ответа и инкрементировать пузомерку рейтинг - зарегистрироваться нужно.

Теперь о том, чем он полезен. Я увидел в нём следующие фишки, которые мне очень понравились.

А. Логические задачки, заставляющие мозг скрипеть, а руки - терзать карандаш и переводить бумагу. Много задач. На логику и математику. Регулярное их решение продлевает жизнь и нефигово "затачивает пилу" логического мышления.

Б. Информатика. Более трех сотен программистских задач. На арифметику, на умение составлять алгоритмы или реализовывать их на языке программирования. Много задач взято с проекта "Эйлер". Что касается выбора средств - тут полная свобода (лично я пользуюсь Python). Нужен только ответ.
Часть задачек решается тупо перебором. Но для некоторых перебор - слишком дорогое удовольствие: задача будет считаться очень, очень долго. Вот тут приходится включать мозг, оптимизировать алгоритм, внимательнее читать условия задачи, применять хитрости, присущие языку, на котором эта задача решается.
Собственно, ради этого всё и затевается. Хочешь - жди неделями, когда найдётся ответ, а хочешь - задумайся, пересмотри алгоритм и может так получиться, что ждать придется не больше минуты. Если есть математическое образование - то да, ждать вообще не придётся ;-)
Ещё я заметил, что с помощью "Диофанта" я воспитываю в себе очень хорошую привычку. А именно, получая некий расчётный ответ в какой-либо задачке, я должен быть уверен на 100% в его правильности. Диофант ошибки не прощает, что сказывается на рейтинге. Поэтому приходится внимательно читать условия и проводить дополнительные проверки.

Резюме такое. Ресурс очень полезный. В качестве источника задач при изучении нового языка программирования и просто для разминки мозгов.

Отчёты отчётами, но "пила" всегда должна быть острой!