Домой Edit me on GitHub

2020-12-05

Каналы передачи данных | Сетевое программирование | Базы данных | Основы Веб-программирования

Paste

Примечание

Исходный код доступен по адресу:

https://github.com/iitwebdev/lectures_wsgi_example

Python Paste, или просто Paste — набор программ для веб-разработки. Включает в себя множество различных middleware, WSGI-сервер и другое. Был разработан Яном Бикингом, чтобы показать всю красоту спецификации WSGI, которая на тот момент была еще в черновиках. Проект больше академический и до недавнего времени не имел даже поддержки Python 3, но несмотря на это, многие современные фреймворки взяли за основу примеры из Paste (TurboGears, Zope, Pylons, Pyramid)

В нем есть готовая поддержка самых разных способов аутентификации (Basic, Digest, form, signed cookie, auth_tkt), поддержка корректной и удобной генерации ответов и заголовков (к примеру редиректы, Cache-control, Expires, gzipper и прочие). Различные базовые средства комбинации приложений (URLMap, Cascade, Recursive), статических данных (с учетом Etag, If-Modified итп).

Некоторые возможности paste мы рассмотрели в разделе WSGI (pep-333).

Предупреждение

Примеры работают только в Python3

HTTP server

Встроенный WSGI сервер wsgiref появился в Python начиная с версии 2.5, частично вобрав наработки из модуля paste.httpserver. На данный момент целесообразно использовать встроенный в Python модуль wsgiref или сторонние более производительные реализации waitress и gunicorn.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def blog(environ, start_response):
    start_response(
        '200 OK',
        [('Content-Type', 'text/plain')]
    )
    return [b'Simple Blog', ]


if __name__ == '__main__':
    from wsgiref.simple_server import make_server
    httpd = make_server('0.0.0.0', 8000, blog)
    httpd.serve_forever()

В нашем случае подойдет любая реализация сервера отвечающего стандарту WSGI, поэтому в примерах используется paste.httpserver. С его помощью запустим простое WSGI-приложение:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def blog(environ, start_response):
    start_response(
        '200 OK',
        [('Content-Type', 'text/plain')]
    )
    return [b'Simple Blog', ]


if __name__ == '__main__':
    from paste.httpserver import serve
    serve(blog, host='0.0.0.0', port=8000)
../../_images/1_0_step_dia.svg

Схема работы WSGI-приложения Blog

Теперь приложение доступно по адресу http://localhost:8000/.

../../_images/1_0_step.png

Главная страница блога

Примечание

Стоит отметить, что приложение будет доступно по любому пути этого адреса, например:

URL диспетчеризация

Доступ к WSGI приложению обычно осуществляется по конкретным URL адресам (ресурсам). В нашем примере приложение blog должно быть доступно только по корневому URL адресу, иные адреса должны выдавать страницу с ошибкой 404.

Для разделения путей напишем WSGI-middleware URLDispatch.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
class URLDispatch(object):

    def __init__(self, app_list):
        self.app_list = app_list

    def __call__(self, environ, start_response):
        path_info = environ.get('PATH_INFO', '')
        for prefix, app in self.app_list:
            if path_info == prefix or path_info == prefix+'/':
                return app(environ, start_response)
        start_response('404 Not Found',
                       [('content-type', 'text/plain')])
        return [b'not found']
../../_images/1_1_step_dia.svg

URLDispatch middleware

Добавим настройки в наше приложение:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from middlewares.urldispatch import URLDispatch


def blog(environ, start_response):
    start_response(
        '200 OK',
        [('Content-Type', 'text/plain')]
    )
    return [b'Simple Blog', ]


# URL dispatching middleware
app_list = [
    ('/', blog),
]
dispatch = URLDispatch(app_list)

if __name__ == '__main__':
    from paste.httpserver import serve
    serve(dispatch, host='0.0.0.0', port=8000)

Любой путь, отличающийся от корневого (http://localhost:8000/), по которому доступно приложение blog, будет инициализировать код ошибки 404.

../../_images/1_1_step.png

404 Not Found

Такой механизм в Веб-разработке называется URL маршрутизация или диспетчеризация, более подробно об этом будет говориться в разделе Маршруты.

Структура нашего блога будет состоять из следующих страниц:

Страницы блога
Название URL Описание
Главная / Показывает все записи в блоге, отсортированные по дате
(CREATE) Добавление /article/add Форма добавления новой статьи
(READ) Просмотр /article/{id} Показывает конкретную статью, соответствующую {id}
(UPDATE) Редактирование /article/{id}/edit Редактирование статьи по {id}
(DELETE) Удаление /article/{id}/delete Удаление статьи по {id}

По сути блог является стандартным CRUD (CREATE-READ-UPDATE-DELETE) интерфейсом, каждую часть которого будет реализовывать свое отдельное WSGI приложение, связанное со своим URL адресом.

../../_images/1_2_step_dia.svg

Сопоставление путей и WSGI-приложений

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from middlewares.urldispatch import URLDispatch


class BaseBlog(object):

    def __init__(self, environ, start_response):
        self.environ = environ
        self.start = start_response


class BlogIndex(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog'


class BlogCreate(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> CREATE'


class BlogRead(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> READ'


class BlogUpdate(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> UPDATE'


class BlogDelete(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> DELETE'


# URL dispatching middleware
app_list = [
    ('/', BlogIndex),
    ('/article/add', BlogCreate),
    ('/article/{id}', BlogRead),
    ('/article/{id}/edit', BlogUpdate),
    ('/article/{id}/delete', BlogDelete),
]
dispatch = URLDispatch(app_list)

if __name__ == '__main__':
    from paste.httpserver import serve
    serve(dispatch, host='0.0.0.0', port=8000)

Примечание

Обратите внимание, что адреса доступны по следующим ссылкам:

Если вместо {id} подставить цифру, то вернется 404 ошибка.

Пока наша реализация роутов ничего не знает про символы типа «{id}», поэтому мы не можем заменять их числом. Чтобы это исправить научим URLDispatch middleware понимать регулярные выражения.

Страницы блога
Название URL Описание
Главная / Показывает все записи в блоге, отсортированные по дате
(CREATE) Добавление /article/add Форма добавления новой статьи
(READ) Просмотр ^/article/(?P<id>d+)/$ Показывает конкретную статью, соответствующую {id}
(UPDATE) Редактирование ^/article/(?P<id>d+)/edit$ Редактирование статьи по {id}
(DELETE) Удаление ^/article/(?P<id>d+)/delete$ Удаление статьи по {id}
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
class RegexDispatch(object):

    def __init__(self, app_list):
        self.app_list = app_list

    def __call__(self, environ, start_response):
        path_info = environ.get('PATH_INFO', '')
        for prefix, app in self.app_list:
            if path_info == prefix or path_info == prefix+'/':
                return app(environ, start_response)
            match = re.match(prefix, path_info) or\
                re.match(prefix, path_info+'/')
            if match and match.groupdict():
                environ['url_params'] = match.groupdict()
                return app(environ, start_response)
        start_response('404 Not Found', [('content-type', 'text/plain')])
        return [b'not found']
../../_images/1_3_step_dia.svg

URL пути на регулярных выражениях

Подменим символы «{id}» в адресах на регулярные выражения:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from middlewares.urldispatch import RegexDispatch


class BaseBlog(object):

    def __init__(self, environ, start_response):
        self.environ = environ
        self.start = start_response


class BlogIndex(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog'


class BlogCreate(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> CREATE'


class BlogRead(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> READ'


class BlogUpdate(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> UPDATE'


class BlogDelete(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> DELETE'


# URL dispatching middleware
app_list = [
    ('/', BlogIndex),
    ('/article/add', BlogCreate),
    (r'^/article/(?P<id>\d+)/$', BlogRead),
    (r'^/article/(?P<id>\d+)/edit/$', BlogUpdate),
    (r'^/article/(?P<id>\d+)/delete/$', BlogDelete),
]
dispatch = RegexDispatch(app_list)

if __name__ == '__main__':
    from paste.httpserver import serve
    serve(dispatch, host='0.0.0.0', port=8000)

Примечание

Теперь можно переходить по URL’ам с числами вместо {id}, например:

Данные

Приложения обычно представляют данные, которые собирают из разных мест (БД, память, файлы, и т.д.), для блога такими данными как раз являются статьи и возможно комментарии к ним.

Оформим данные в виде списка, каждая запись которого будет являться статьей со следующими ключами id, title, content.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
ARTICLES = [
    {'id': 1, 'title': 'Lorem ipsum dolor sid amet!',
     'content': '''Lorem ipsum dolor sit amet, consectetur adipiscing elit.
     Curabitur vel tortor eleifend, sollicitudin nisl quis, lacinia augue.
     Duis quam est, laoreet sit amet justo vitae, viverra egestas sem.
     Maecenas pellentesque augue in nibh feugiat tincidunt. Nunc magna ante,
     mollis vitae ultricies eu, consectetur id ante. In ut libero eleifend,
     blandit ipsum a, ullamcorper nunc. Sed bibendum eget odio eget
     pellentesque. Curabitur elit felis, pellentesque id feugiat et, tincidunt
     ut mauris. Integer vitae vehicula nunc. Integer ullamcorper, nunc in
     volutpat auctor, elit leo convallis nulla, vitae varius mi nisl ac lorem.
     Sed a lacus mi. In hac habitasse platea dictumst. Cras in posuere velit,
     id dignissim nisl. Interdum et malesuada fames ac ante ipsum primis in
     faucibus. Nulla bibendum suscipit convallis.'''},
    {'id': 2, 'title': 'Hello', 'content': 'Test2'},
    {'id': 3, 'title': 'World', 'content': 'Test2'}, ]

Главная страница формируется перебором статей в списке:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
class BlogIndex(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1>Simple Blog</h1>'
        for article in ARTICLES:
            yield str.encode(
                '''
                {0} - <a href="/article/{0}">{1}</a>
                (<a href="/article/{0}/delete">delete</a>)<br/>
                '''.format(
                    article['id'],
                    article['title']
                )
            )
../../_images/1_4_step.png

Список статей на главной странице

WSGI-приложения BlogRead, BlogUpdate и BlogDelete теперь наследуются от специально класса BaseArticle, он берет id статьи (переменная окружения, которую добавляет middlwware RegexDispatch) и находит ее среди списка данных.

1
2
3
4
5
6
7
8
9
class BaseArticle(BaseBlog):

    def __init__(self, *args):
        super(BaseArticle, self).__init__(*args)
        article_id = self.environ['url_params']['id']
        (self.index,
         self.article) = next(((i, art) for i, art in enumerate(ARTICLES)
                               if art['id'] == int(article_id)),
                              (None, None))

Приложение BlogRead, отвечающее за чтение статьи, выводит его содержимое или отдает 404 ошибку:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
class BlogRead(BaseArticle):

    def __iter__(self):
        if not self.article:
            self.start('404 Not Found', [('content-type', 'text/plain')])
            yield b'not found'
            return

        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> READ</h1>'
        yield str.encode('<h2>{}</h2>'.format(self.article['title']))
        yield str.encode(self.article['content'])
../../_images/1_4_step2.png

Страница статьи

Приложение, удаляющее статью — BlogDelete, удаляет объект из списка данных и возвращает статус ответа 302 Fount с заголовком Location: /, указывающий браузеру, что нужно перейти на главную страницу (перенаправление).

1
2
3
4
5
6
7
8
class BlogDelete(BaseArticle):

    def __iter__(self):
        self.start('302 Found',  # '301 Moved Permanently',
                   [('Content-Type', 'text/html'),
                    ('Location', '/')])
        ARTICLES.pop(self.index)
        yield b''

Полный код с изменениями:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
from middlewares.urldispatch import RegexDispatch

ARTICLES = [
    {'id': 1, 'title': 'Lorem ipsum dolor sid amet!',
     'content': '''Lorem ipsum dolor sit amet, consectetur adipiscing elit.
     Curabitur vel tortor eleifend, sollicitudin nisl quis, lacinia augue.
     Duis quam est, laoreet sit amet justo vitae, viverra egestas sem.
     Maecenas pellentesque augue in nibh feugiat tincidunt. Nunc magna ante,
     mollis vitae ultricies eu, consectetur id ante. In ut libero eleifend,
     blandit ipsum a, ullamcorper nunc. Sed bibendum eget odio eget
     pellentesque. Curabitur elit felis, pellentesque id feugiat et, tincidunt
     ut mauris. Integer vitae vehicula nunc. Integer ullamcorper, nunc in
     volutpat auctor, elit leo convallis nulla, vitae varius mi nisl ac lorem.
     Sed a lacus mi. In hac habitasse platea dictumst. Cras in posuere velit,
     id dignissim nisl. Interdum et malesuada fames ac ante ipsum primis in
     faucibus. Nulla bibendum suscipit convallis.'''},
    {'id': 2, 'title': 'Hello', 'content': 'Test2'},
    {'id': 3, 'title': 'World', 'content': 'Test2'}, ]


class BaseBlog(object):

    def __init__(self, environ, start_response):
        self.environ = environ
        self.start = start_response


class BaseArticle(BaseBlog):

    def __init__(self, *args):
        super(BaseArticle, self).__init__(*args)
        article_id = self.environ['url_params']['id']
        (self.index,
         self.article) = next(((i, art) for i, art in enumerate(ARTICLES)
                               if art['id'] == int(article_id)),
                              (None, None))


class BlogIndex(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1>Simple Blog</h1>'
        for article in ARTICLES:
            yield str.encode(
                '''
                {0} - <a href="/article/{0}">{1}</a>
                (<a href="/article/{0}/delete">delete</a>)<br/>
                '''.format(
                    article['id'],
                    article['title']
                )
            )


class BlogCreate(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> CREATE'


class BlogRead(BaseArticle):

    def __iter__(self):
        if not self.article:
            self.start('404 Not Found', [('content-type', 'text/plain')])
            yield b'not found'
            return

        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> READ</h1>'
        yield str.encode('<h2>{}</h2>'.format(self.article['title']))
        yield str.encode(self.article['content'])


class BlogUpdate(BaseArticle):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/plain')])
        yield b'Simple Blog -> UPDATE'


class BlogDelete(BaseArticle):

    def __iter__(self):
        self.start('302 Found',  # '301 Moved Permanently',
                   [('Content-Type', 'text/html'),
                    ('Location', '/')])
        ARTICLES.pop(self.index)
        yield b''


# URL dispatching middleware
app_list = [
    ('/', BlogIndex),
    ('/article/add', BlogCreate),
    (r'^/article/(?P<id>\d+)/$', BlogRead),
    (r'^/article/(?P<id>\d+)/edit/$', BlogUpdate),
    (r'^/article/(?P<id>\d+)/delete/$', BlogDelete),
]
dispatch = RegexDispatch(app_list)

if __name__ == '__main__':
    from paste.httpserver import serve
    serve(dispatch, host='0.0.0.0', port=8000)

Формы

Для создания статьи требуется HTML форма, где указываются заголовок и содержание. В WSGI приложении BlogCreate, запрос с методом GET возвращает HTML форму, а POST записывает данные в список ARTICLES, после чего перенаправляет на главную страницу.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class BlogCreate(BaseBlog):

    def __iter__(self):
        if self.environ['REQUEST_METHOD'].upper() == 'POST':
            from urllib.parse import parse_qs
            values = parse_qs(self.environ['wsgi.input'].read())
            max_id = max([art['id'] for art in ARTICLES])
            ARTICLES.append(
                {'id': max_id+1,
                 'title': values[b'title'].pop().decode(),
                 'content': values[b'content'].pop().decode()
                 }
            )
            self.start('302 Found',
                       [('Content-Type', 'text/html'),
                        ('Location', '/')])
            return

        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> CREATE</h1>'
        yield b'''
<form action="" method="POST">
    Title:<br>
    <input type="text" name="title"><br>
    Content:<br>
    <textarea name="content"></textarea><br><br>
    <input type="submit" value="Submit">
</form>'''
../../_images/1_5_step.png

Форма создания новой статьи

Обновление статей происходит схожим образом, за исключением того, что в форму подставляются уже существующие значения и вместо добавления нового объекта в список ARTICLES, обновляется уже существующий.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class BlogUpdate(BaseArticle):

    def __iter__(self):
        if self.environ['REQUEST_METHOD'].upper() == 'POST':
            from urllib.parse import parse_qs
            values = parse_qs(self.environ['wsgi.input'].read())
            self.article['title'] = values[b'title'].pop().decode()
            self.article['content'] = values[b'content'].pop().decode()
            self.start('302 Found',
                       [('Content-Type', 'text/html'),
                        ('Location', '/')])
            return
        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> UPDATE</h1>'
        yield str.encode(
            '''
            <form action="" method="POST">
                Title:<br>
                <input type="text" name="title" value="{0}"><br>
                Content:<br>
                <textarea name="content">{1}</textarea><br><br>
                <input type="submit" value="Submit">
            </form>
            '''.format(
                self.article['title'],
                self.article['content']
            )
        )
../../_images/1_5_step2.png

Форма редактирования статьи

Полный код с изменениями:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
# third-party
from middlewares.urldispatch import RegexDispatch

ARTICLES = [
    {'id': 1, 'title': 'Lorem ipsum dolor sid amet!',
     'content': '''Lorem ipsum dolor sit amet, consectetur adipiscing elit.
     Curabitur vel tortor eleifend, sollicitudin nisl quis, lacinia augue.
     Duis quam est, laoreet sit amet justo vitae, viverra egestas sem.
     Maecenas pellentesque augue in nibh feugiat tincidunt. Nunc magna ante,
     mollis vitae ultricies eu, consectetur id ante. In ut libero eleifend,
     blandit ipsum a, ullamcorper nunc. Sed bibendum eget odio eget
     pellentesque. Curabitur elit felis, pellentesque id feugiat et, tincidunt
     ut mauris. Integer vitae vehicula nunc. Integer ullamcorper, nunc in
     volutpat auctor, elit leo convallis nulla, vitae varius mi nisl ac lorem.
     Sed a lacus mi. In hac habitasse platea dictumst. Cras in posuere velit,
     id dignissim nisl. Interdum et malesuada fames ac ante ipsum primis in
     faucibus. Nulla bibendum suscipit convallis.'''},
    {'id': 2, 'title': 'Hello', 'content': 'Test2'},
    {'id': 3, 'title': 'World', 'content': 'Test2'}, ]


class BaseBlog(object):

    def __init__(self, environ, start_response):
        self.environ = environ
        self.start = start_response


class BaseArticle(BaseBlog):

    def __init__(self, *args):
        super(BaseArticle, self).__init__(*args)
        article_id = self.environ['url_params']['id']
        (self.index,
         self.article) = next(((i, art) for i, art in enumerate(ARTICLES)
                               if art['id'] == int(article_id)),
                              (None, None))


class BlogIndex(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1>Simple Blog</h1>'
        yield b'<a href="/article/add">Add article</a>'
        yield b'<br />'
        yield b'<br />'
        for article in ARTICLES:
            yield str.encode(
                '''
                {0} - (<a href="/article/{0}/delete">delete</a> |
                <a href="/article/{0}/edit">edit</a>)
                <a href="/article/{0}">{1}</a><br />
                '''.format(
                    article['id'],
                    article['title']
                )
            )


class BlogCreate(BaseBlog):

    def __iter__(self):
        if self.environ['REQUEST_METHOD'].upper() == 'POST':
            from urllib.parse import parse_qs
            values = parse_qs(self.environ['wsgi.input'].read())
            max_id = max([art['id'] for art in ARTICLES])
            ARTICLES.append(
                {'id': max_id+1,
                 'title': values[b'title'].pop().decode(),
                 'content': values[b'content'].pop().decode()
                 }
            )
            self.start('302 Found',
                       [('Content-Type', 'text/html'),
                        ('Location', '/')])
            return

        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> CREATE</h1>'
        yield b'''
<form action="" method="POST">
    Title:<br>
    <input type="text" name="title"><br>
    Content:<br>
    <textarea name="content"></textarea><br><br>
    <input type="submit" value="Submit">
</form>'''


class BlogRead(BaseArticle):

    def __iter__(self):
        if not self.article:
            self.start('404 Not Found', [('content-type', 'text/plain')])
            yield b'not found'
            return

        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> READ</h1>'
        yield str.encode('<h2>{}</h2>'.format(self.article['title']))
        yield str.encode(self.article['content'])


class BlogUpdate(BaseArticle):

    def __iter__(self):
        if self.environ['REQUEST_METHOD'].upper() == 'POST':
            from urllib.parse import parse_qs
            values = parse_qs(self.environ['wsgi.input'].read())
            self.article['title'] = values[b'title'].pop().decode()
            self.article['content'] = values[b'content'].pop().decode()
            self.start('302 Found',
                       [('Content-Type', 'text/html'),
                        ('Location', '/')])
            return
        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> UPDATE</h1>'
        yield str.encode(
            '''
            <form action="" method="POST">
                Title:<br>
                <input type="text" name="title" value="{0}"><br>
                Content:<br>
                <textarea name="content">{1}</textarea><br><br>
                <input type="submit" value="Submit">
            </form>
            '''.format(
                self.article['title'],
                self.article['content']
            )
        )


class BlogDelete(BaseArticle):

    def __iter__(self):
        self.start('302 Found',  # '301 Moved Permanently',
                   [('Content-Type', 'text/html'),
                    ('Location', '/')])
        ARTICLES.pop(self.index)
        yield b''


# URL dispatching middleware
app_list = [
    ('/', BlogIndex),
    ('/article/add', BlogCreate),
    (r'^/article/(?P<id>\d+)/$', BlogRead),
    (r'^/article/(?P<id>\d+)/edit/$', BlogUpdate),
    (r'^/article/(?P<id>\d+)/delete/$', BlogDelete),
]
dispatch = RegexDispatch(app_list)

if __name__ == '__main__':
    from paste.httpserver import serve
    serve(dispatch, host='0.0.0.0', port=8000)

Авторизация

Авторизация поможет защитить ресурсы от сторонних пользователей, в первую очередь это касается операций, которые изменяют данные (BlogCreate, BlogUpdate, BlogDelete). В эти WSGI приложения необходимо будет добавить проверку пользователя.

../../_images/1_6_step_dia.svg

BasicAuth WSGI-middleware для авторизации

В нашем примере используется алгоритм BasicAuth и WSGI-middleware middlewares.basicauth.BasicAuth.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class BasicAuth(object):

    def __init__(self, app, users, realm='www'):
        self.app = app
        self.users = users
        self.realm = realm

    def __call__(self, environ, start_response):
        auth = environ.get('HTTP_AUTHORIZATION')
        if not auth:
            return self.auth_required(environ, start_response)
        auth_type, enc_auth_info = auth.split(None, 1)
        assert auth_type.lower() == 'basic'
        auth_info = base64.b64decode(enc_auth_info)
        username, password = auth_info.decode().split(':', 1)
        if self.users.get(username) != password:
            return self.auth_required(environ, start_response)
        environ['REMOTE_USER'] = username
        return self.app(environ, start_response)

    def auth_required(self, environ, start_response):
        status = '401 Authorization Required'
        headers = [
            ('content-type', 'text/plain'),
            ('WWW-Authenticate', 'Basic realm="%s"' % self.realm)
        ]
        start_response(status, headers)
        return [b'authentication required']

Полный код с изменениями:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
from middlewares.urldispatch import RegexDispatch
from middlewares.basicauth import BasicAuth

ARTICLES = [
    {'id': 1, 'title': 'Lorem ipsum dolor sid amet!',
     'content': '''Lorem ipsum dolor sit amet, consectetur adipiscing elit.
     Curabitur vel tortor eleifend, sollicitudin nisl quis, lacinia augue.
     Duis quam est, laoreet sit amet justo vitae, viverra egestas sem.
     Maecenas pellentesque augue in nibh feugiat tincidunt. Nunc magna ante,
     mollis vitae ultricies eu, consectetur id ante. In ut libero eleifend,
     blandit ipsum a, ullamcorper nunc. Sed bibendum eget odio eget
     pellentesque. Curabitur elit felis, pellentesque id feugiat et, tincidunt
     ut mauris. Integer vitae vehicula nunc. Integer ullamcorper, nunc in
     volutpat auctor, elit leo convallis nulla, vitae varius mi nisl ac lorem.
     Sed a lacus mi. In hac habitasse platea dictumst. Cras in posuere velit,
     id dignissim nisl. Interdum et malesuada fames ac ante ipsum primis in
     faucibus. Nulla bibendum suscipit convallis.'''},
    {'id': 2, 'title': 'Hello', 'content': 'Test2'},
    {'id': 3, 'title': 'World', 'content': 'Test2'}, ]


class BaseBlog(object):

    def __init__(self, environ, start_response):
        self.environ = environ
        self.start = start_response


class BaseArticle(BaseBlog):

    def __init__(self, *args):
        super(BaseArticle, self).__init__(*args)
        article_id = self.environ['url_params']['id']
        (self.index,
         self.article) = next(((i, art) for i, art in enumerate(ARTICLES)
                               if art['id'] == int(article_id)),
                              (None, None))


class BlogIndex(BaseBlog):

    def __iter__(self):
        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1>Simple Blog</h1>'
        yield b'<a href="/article/add">Add article</a>'
        yield b'<br />'
        yield b'<br />'
        for article in ARTICLES:
            yield str.encode(
                '''
                {0} - (<a href="/article/{0}/delete">delete</a> |
                <a href="/article/{0}/edit">edit</a>)
                <a href="/article/{0}">{1}</a><br />
                '''.format(
                    article['id'],
                    article['title']
                )
            )


class BlogCreate(BaseBlog):

    def __iter__(self):
        if self.environ['REQUEST_METHOD'].upper() == 'POST':
            from urllib.parse import parse_qs
            values = parse_qs(self.environ['wsgi.input'].read())
            max_id = max([art['id'] for art in ARTICLES])
            ARTICLES.append(
                {'id': max_id+1,
                 'title': values[b'title'].pop().decode(),
                 'content': values[b'content'].pop().decode()
                 }
            )
            self.start('302 Found',
                       [('Content-Type', 'text/html'),
                        ('Location', '/')])
            return

        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> CREATE</h1>'
        yield b'''
            <form action="" method="POST">
                Title:<br>
                <input type="text" name="title"><br>
                Content:<br>
                <textarea name="content"></textarea><br><br>
                <input type="submit" value="Submit">
            </form>
            '''


class BlogRead(BaseArticle):

    def __iter__(self):
        if not self.article:
            self.start('404 Not Found', [('content-type', 'text/plain')])
            yield b'not found'
            return

        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> READ</h1>'
        yield str.encode('<h2>{}</h2>'.format(self.article['title']))
        yield str.encode(self.article['content'])


class BlogUpdate(BaseArticle):

    def __iter__(self):
        if self.environ['REQUEST_METHOD'].upper() == 'POST':
            from urllib.parse import parse_qs
            values = parse_qs(self.environ['wsgi.input'].read())
            self.article['title'] = values[b'title'].pop().decode()
            self.article['content'] = values[b'content'].pop().decode()
            self.start('302 Found',
                       [('Content-Type', 'text/html'),
                        ('Location', '/')])
            return
        self.start('200 OK', [('Content-Type', 'text/html')])
        yield b'<h1><a href="/">Simple Blog</a> -> UPDATE</h1>'
        yield str.encode(
            '''
            <form action="" method="POST">
                Title:<br>
                <input type="text" name="title" value="{0}"><br>
                Content:<br>
                <textarea name="content">{1}</textarea><br><br>
                <input type="submit" value="Submit">
            </form>
            '''.format(
                self.article['title'],
                self.article['content']
            )
        )


class BlogDelete(BaseArticle):

    def __iter__(self):
        self.start('302 Found',  # '301 Moved Permanently',
                   [('Content-Type', 'text/html'),
                    ('Location', '/')])
        ARTICLES.pop(self.index)
        yield b''


# BasicAuth applications
passwd = {'admin': '123'}
create = BasicAuth(BlogCreate, passwd)
update = BasicAuth(BlogUpdate, passwd)
delete = BasicAuth(BlogDelete, passwd)

# URL dispatching middleware
app_list = [
    ('/', BlogIndex),
    ('/article/add', create),
    (r'^/article/(?P<id>\d+)/$', BlogRead),
    (r'^/article/(?P<id>\d+)/edit/$', update),
    (r'^/article/(?P<id>\d+)/delete/$', delete),
]
dispatch = RegexDispatch(app_list)

if __name__ == '__main__':
    from paste.httpserver import serve
    serve(dispatch, host='0.0.0.0', port=8000)
Previous: Веб-программирование Next: Разделение кода