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
Встроенный 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) |
Теперь приложение доступно по адресу http://localhost:8000/.
Примечание
Стоит отметить, что приложение будет доступно по любому пути этого адреса, например:
Доступ к 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'] |
Добавим настройки в наше приложение:
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.
Такой механизм в Веб-разработке называется 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 адресом.
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'] |
Подменим символы «{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']
)
) |
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']) |
Приложение, удаляющее статью — 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>''' |
Обновление статей происходит схожим образом, за исключением того, что в форму
подставляются уже существующие значения и вместо добавления нового объекта в
список 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']
)
) |
Полный код с изменениями:
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 приложения необходимо будет
добавить проверку пользователя.
В нашем примере используется алгоритм 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) |