Thoughts and Practices: April 2013

Monday, April 22, 2013

Simple-TODO Pyramid实现版

看到@python4cn 网站上有个专栏列了各个python框架的SimpleTODO实现，唯独没有Pyramid，就随手写了一个，略做补充。具体的项目地址在 https://github.com/eryxlee/pyramid_koans/tree/master/Simple-TODO

import os
import datetime
import transaction

from pyramid.config import Configurator
from pyramid.view import view_config
from pyramid.httpexceptions import HTTPFound

from sqlalchemy import engine_from_config

from sqlalchemy import (
    Column,
    Integer,
    DateTime,
    Unicode,
    )

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import (
    scoped_session,
    sessionmaker,
    )

from zope.sqlalchemy import ZopeTransactionExtension

DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
Base = declarative_base()


class Todo(Base):
    __tablename__ = 'todo'

    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(Unicode(255), default=u'')
    post_date = Column(DateTime, default=datetime.datetime.now)
    finished = Column(Integer, default=0)

    def __init__(self, title, post_date, finished):
        self.title = title
        self.post_date = post_date
        self.finished = finished


@view_config(route_name='index',  renderer='/index.mako')
def index_view(request):
    todos = DBSession.query(Todo).order_by(Todo.post_date.desc()).all()
    return {'todos': todos}


@view_config(route_name='add', request_method='POST')
def add_post_view(request):
    title = request.params.get('title', '')
    with transaction.manager:
        todo = Todo(title=title, post_date=datetime.datetime.now(), finished=0)
        DBSession.add(todo)

    raise HTTPFound(location = request.route_url('index'))


@view_config(route_name='edit', request_method='GET', renderer='/edit.mako')
def edit_view(request):
    id = int(request.matchdict.get('id'))
    todo = DBSession.query(Todo).filter(Todo.id==id).first()

    return {'todo': todo}


@view_config(route_name='edit', request_method='POST')
def edit_post_view(request):
    id = int(request.matchdict.get('id'))
    title = request.params.get('title')
    with transaction.manager:
        DBSession.query(Todo).filter(Todo.id==id).update({Todo.title:title})

    raise HTTPFound(location = request.route_url('index'))


@view_config(route_name='finish')
def finish_view(request):
    id = int(request.matchdict.get('id'))
    status = request.params.get('status', 'yes')
    finished = {'yes':1, 'no':0}.get(status)
    if finished != None:
        with transaction.manager:
            DBSession.query(Todo).filter(Todo.id==id).update({Todo.finished:finished})

    raise HTTPFound(location = request.route_url('index'))


@view_config(route_name='delete')
def delete_view(request):
    id = int(request.matchdict.get('id'))
    with transaction.manager:
        DBSession.query(Todo).filter(Todo.id==id).delete()

    raise HTTPFound(location = request.route_url('index'))


def my_static_path(self, path, **kw):
    if not os.path.isabs(path):
        if not ':' in path:
            path = '%s:%s/%s' % ("simpletodo", 'static', path)
    kw['_app_url'] = self.script_name

    return self.static_url(path, **kw)


def main(global_config, **settings):
    engine = engine_from_config(settings, 'sqlalchemy.')
    DBSession.configure(bind=engine)
    Base.metadata.bind = engine
    config = Configurator(settings=settings)
    config.add_request_method(my_static_path, 'static_path')
    config.add_static_view('static', 'static', cache_max_age=3600)
    config.add_route('index', '/')
    config.add_route('add', '/todo/new')
    config.add_route('edit', '/todo/{id:\d+}/edit')
    config.add_route('finish', '/todo/{id:\d+}/finish')
    config.add_route('delete', '/todo/{id:\d+}/delete')
    config.scan()

    return config.make_wsgi_app()

Thursday, April 18, 2013

Python Requests 使用摘要二

五. 带参数访问

1. 普通参数

>>> payload = {'key1': 'value1', 'key2': 'value2'}

>>> r = requests.get("http://httpbin.org/get", params=payload)

>>> r.url

u'http://httpbin.org/get?key2=value2&key1=value1'

>>> r.text

u'{\n "url": "http://httpbin.org/get?key2=value2&key1=value1",\n }'

2. 带中文参数

>>> payload = {'key1': 'value1', 'key2': u'中文'}

>>> r = requests.get("http://httpbin.org/get", params=payload)

>>> r.url

u'http://httpbin.org/get?key2=%E4%B8%AD%E6%96%87&key1=value1'

>>> r.text

u'{\n "url": "http://httpbin.org/get?key2=%E4%B8%AD%E6%96%87&key1=value1",\n }'

3. json格式

>>> import json

>>> url = 'https://api.github.com/some/endpoint'

>>> payload = {'some': 'data'}

>>> headers = {'content-type': 'application/json'}

>>> r = requests.post(url, data=json.dumps(payload), headers=headers)

六. 文件操作

1. 文件下载

r = requests.get('http://img4.cache.netease.com/travel/2013/4/7/2013040718512699794.jpg')

from PIL import Image

from StringIO import StringIO

i = Image.open(StringIO(r.content))

i.save('1.jpg')

2. 文件上传

>>> url = 'http://httpbin.org/post'

>>> files = {'file': open('report.xls', 'rb')}

>>> r = requests.post(url, files=files)

>>> r.text

3. 上传时指定文件名

>>> url = 'http://httpbin.org/post'

>>> files = {'file': ('report.xls', open('report.xls', 'rb'))}

>>> r = requests.post(url, files=files)

>>> r.text

4. 按文件接收字符串

>>> url = 'http://httpbin.org/post'

>>> files = {'file': ('report.csv', 'some,data,to,send\nanother,row,to,send\n')}

>>> r = requests.post(url, files=files)

>>> r.text

5. 流式上传，大文件上传时，不需要全部装载到内存

with open('massive-body') as f:

requests.post('http://some.url/streamed', data=f)

6. 流式下载

r = requests.post('https://stream.twitter.com/1/statuses/filter.json',

data={'track': 'requests'}, auth=('username', 'password'), stream=True)

for line in r.iter_lines():

if line: # filter out keep-alive new lines

print json.loads(line)

七. 代理设置

1. 代理参数

proxies = {

"http": "http://10.10.1.10:3128",

"https": "http://10.10.1.10:1080",

}

requests.get("http://example.org", proxies=proxies)

>>> r = requests.get('http://ifconfig.me/ip')

>>> r.text

u'116.226.xx.xxx\n'

>>> proxies = {

... "http": "http://175.136.xxx.xx",

... }

>>> r = requests.get("http://ifconfig.me/ip", proxies=proxies)

>>> r.text

u'175.136.xxx.xx\n'

2. 环境变量

$ export HTTP_PROXY="http://10.10.1.10:3128"

$ export HTTPS_PROXY="http://10.10.1.10:1080"

$ python

>>> import requests

>>> requests.get("http://example.org")

3. 如果代理授权方式用的是HTTP Basic Auth，则可以

proxies = {

"http": "http://user:pass@10.10.1.10:3128/",

}

八 Session

如果需要在多次访问之间保持状态，则需要用到requests中的Session对象。

>>> s = requests.Session()

>>> s.get('http://httpbin.org/cookies/set/sessioncookie/123456789')

>>> r = s.get("http://httpbin.org/cookies")

>>> r.text

u'{\n "cookies": {\n "sessioncookie": "123456789"\n }\n}'

另外，Session支持Keep-Alive，在同一个session中的多个请求会自动重用连接。注意，只有所有body内容读完，连接才会放回连接池重用。在使用流式文件的时候小心。

九. 异常出错

1. 网络错误（如DNS出错，链接拒绝等），抛出ConnectionError

2. 遭遇少见的非法HTTP 响应（不是requests能解析的那些404之类异常），抛出HTTPError

3. 超时，抛出Timeout

4. 301之类的跳转次数太多，抛出TooManyRedirects

5. 所有requests抛出的异常均继承自requests.exceptions.RequestException

Python Requests 使用摘要一

Requests是由Kenneth Reitz推出的一个Python HTTP 请求操作包，在使用上，比系统自带的urllib2方便了很多，现在是1.2版本，可以通过easy_install安装。

一. 基本操作

r = requests.get('http://httpbin.org/get')

r = requests.post("http://httpbin.org/post")

r = requests.put("http://httpbin.org/put")

r = requests.delete("http://httpbin.org/delete")

r = requests.head("http://httpbin.org/get")

r = requests.options("http://httpbin.org/get")

print r.headers['allow']

HEAD, OPTIONS, GET

二. 查看返回内容

>>> r = requests.get('http://httpbin.org/get')

1. 响应内容，以 bytes表示

>>> r.content

'{\n "url": "http://httpbin.org/get",\n "headers": {\n "Content-Length": "0",\n "Accept-Encoding": "gzip, deflate, compress",\n "Connection": "close",\n "Accept": "*/*",\n "User-Agent": "python-requests/1.1.0 CPython/2.7.1 Darwin/11.4.2",\n "Host": "httpbin.org"\n },\n "args": {},\n "origin": "..."\n}'

2. 响应内容，以 unicode表示

>>> r.text

u'{\n "url": "http://httpbin.org/get",\n "headers": {\n "Content-Length": "0",\n "Accept-Encoding": "gzip, deflate, compress",\n "Connection": "close",\n "Accept": "*/*",\n "User-Agent": "python-requests/1.1.0 CPython/2.7.1 Darwin/11.4.2",\n "Host": "httpbin.org"\n },\n "args": {},\n "origin": "..."\n}'

3. 友好提示。访问无异常，为True，否则，为False。响应HTTP状态码400及以上均为False。

>>> r.ok

True

4. 访问无异常，返回空，否则，抛出异常

>>> r.raise_for_status()

5. 响应HTTP状态码

>>> r.status_code

200

6. 访问URL

>>> r.url

u'http://httpbin.org/get'

7. 探测到headers里面的charset，要注意这里不是页面上指定的charset

>>> r.encoding

8. 如果headers里面发现不了，则会查找网页内容来探测，不过速度很慢

>>> r.apparent_encoding

9. HTTP Headers内容

>>> r.headers.keys()

['date', 'content-length', 'content-type', 'connection', 'server']

>>>r.headers['Content-Type']

'application/json'

>>> r.headers.get('content-type')

'application/json'

10. Cookies内容

>>> r.cookies.keys()

[]

>>> url = 'http://httpbin.org/cookies'

>>> cookies = dict(cookies_are='working')

>>> r = requests.get(url, cookies=cookies)

>>> r.text

u'{\n "cookies": {\n "cookies_are": "working"\n }\n}'

11. 转化成json格式输出

>>> r.json()

{u'url': u'http://httpbin.org/get', u'headers': {u'Content-Length': u'0', u'Accept-Encoding': u'gzip, deflate, compress', u'Connection': u'close', u'Accept': u'*/*', u'User-Agent': u'python-requests/1.1.0 CPython/2.7.1 Darwin/11.4.2', u'Host': u'httpbin.org'}, u'args': {}, u'origin': u'...'}

12. 发出请求到接到回应的时间差

>>> r.elapsed

datetime.timedelta(0, 1, 695693)

13. iterate内容

>>> a = r.iter_lines()

>>> a

>>> a.next()

>>> a = r.iter_content()

>>> a

>>> a.next()

'<'

>>> a.next()

'!'

>>> a.next()

'D'

>>> a.next()

'O'

>>> a.next()

'C'

14. Timeouts

>>> requests.get('http://github.com', timeout=0.001)

Traceback (most recent call last):

File "", line 1, in

requests.exceptions.Timeout: HTTPConnectionPool(host='github.com', port=80): Request timed out. (timeout=0.001)

三. 访问出错返回的内容

>>> r = requests.get('http://127.0.0.1:6543/abcccc') # 这是一个不存在的链接

1. 响应内容，以 unicode表示

>>> r.text

u'<html>\n <head>\n <title>404 Not Found</title>\n </head>\n <body>\n <h1>404 Not Found</h1>\n The resource could not be found.<br/><br/>\n/abcccc\n\n\n </body>\n</html>'

2. 响应内容，以 bytes表示

>>> r.content

'<html>\n <head>\n <title>404 Not Found</title>\n </head>\n <body>\n <h1>404 Not Found</h1>\n The resource could not be found.<br/><br/>\n/abcccc\n\n\n </body>\n</html>'

3. 友好提示，返回False

>>> r.ok

False

4. 响应HTTP状态码

>>> r.status_code

404

5. 将访问异常外发

>>> r.raise_for_status()

Traceback (most recent call last):

File "", line 1, in

File "/Users/eryxlee/Workshops/python/sandbox/lib/python2.7/site-packages/requests-1.2.0-py2.7.egg/requests/models.py", line 670, in raise_for_status

raise HTTPError(http_error_msg, response=self)

requests.exceptions.HTTPError: 404 Client Error: Not Found

6. 友好提示，返回一个合适阅读的出错提示

>>> r.reason

'Not Found'

四. 跳转情况的返回

>>> r = requests.get('http://github.com')

1. 返回内容

>>> r.status_code

200

>>> r.text[:100]

u'\n\n http://ogp.me/ns#

fb: http://ogp.me/ns/fb# githubog: http'

2. 跳转历史

>>> r.history

(,)

3. 跳转内容

>>> r.history[0].text

u'<html>\r\n<head><title>301 Moved Permanently</title></head>\r\n<body bgcolor="white">\r\n<center><h1>301 Moved Permanently</h1></center>\r\n<hr><center>nginx</center>\r\n</body>\r\n</html>\r\n'

4. 禁止跳转

>>> r = requests.get('http://github.com', allow_redirects=False)

>>> r.status_code

301

>>> r.history

[]

Thoughts and Practices

Monday, April 22, 2013

Simple-TODO Pyramid实现版

Thursday, April 18, 2013

Python Requests 使用摘要二

Python Requests 使用摘要一

About Me

Labels

Blog Archive

Thoughts and Practices

Monday, April 22, 2013

Simple-TODO Pyramid实现版

Thursday, April 18, 2013

Python Requests 使用摘要 二

Python Requests 使用摘要 一

About Me

Subscribe To

Labels

Blog Archive

Python Requests 使用摘要二

Python Requests 使用摘要一