Web 后端

flask 全家桶学习笔记(未完待续)

看到标题有的同学可能就问了,flask 是一个微框架,哪儿来的全家桶啊。其实作为一个框架来说,除非你提供的只有静态页面,那么肯定要和数据库打交道的,肯定是要有后台登录管理以及提供 API 等等一大堆常规工作要做的,这时候就需要各种全家桶组件了,那么这篇文章里介绍的就是 flask + peewee + login + admin + uwsgi 等等一系列的工具。

hello world

from flask import Flask

app = Flask(__name__)

@app.route("/")
def hello():
    return "hello, world"

app.run()

Application Factory Pattern

在前面的例子中,我们都直接在模块中 app = Flask(__name__) 了,这样做实际上是有问题的。官方推荐使用 app factory pattern。

app factory pattern 其实也很简单,就是把 app 的创建包装在了 create_app 函数中,这样做的好处主要有两点:

方便多环境部署

直接导入 app 的话,已经初始化了,无法再更改 app 的配置

from example import app

如果把 app 的创建包装在一个函数中,可以在创建 app 的时候传递不同的参数,可以区分开发测试等不同环境。

def create_app(**kwargs):
    app = Flask(**kwargs)
    return app

from example import create_app
app = create_app(DB_CONN="production")

方便依赖管理

默认情况下,代码可能是这样的,所有的代码都得依赖 app.py

# app.py
app = Flask(__name__)
db = SQLAlchemy(app)

# models.py
from example import db

class User(db.Model):
    pass

使用了 app factory pattern 之后,每个模块都可以不依赖 app.py,而是使用自己的 blueprint

def create_app():
    app = Flask(__name__)
    from example.models import db
    db.init_app(app)
    return app

# models.py
db = SQLAlchemy()

class User(db.Model):
    pass

使用 blueprint

搭建 flask 的后台系统

cookie & session & login

flask 使用了 itsdangerous 库生成和读取 Cookie。flask 默认的 session 也是通过 cookie 实现的。因为 Cookie 是储存在客户端的,所以:

  1. 很难在 session 中存储数据
  2. 每次都会携带 Cookie,影响 HTTP 请求大小
  3. 不需要在服务端有任何存储,使用比较简单

flask-login

flask-login 是 flask 的一个登录框架,它使用了 flask 本身的 session 机制,可以对接各种数据库后端。

初始化 flask-login

from flask_login import LoginManager

manager = LoginManager()
manager.init_app(app)

flask-login 的接口

登录登出

用户登录。使用 loginuser 方法登录后,flasklogin 会设置 cookie。

@app.route("/login")
def login():
    username = request.args.get("username")
    password = request.args.get("password")
    user = User.get(username=username)
    if user.check_password(password):
        login_user(user)

这时候之后的访问就都可以从 session 中加载出用户了。如果需要登出的话:

@app.route("/logout")
@login_required  # 只有登录后可以访问
def logout():
    logout_user()

加载用户

加载用户的接口。这里的 user_id 是直接从 session 中拿到的。

@manager.user_loader
def load_user(user_id):
    return User.get(user_id)

当我们登录完成之后,就可以通过 current_user 这个代理来访问当前用户了。

from flask_login import current_user

其中的 User 类需要提供以下四个方法,不过好在我们可以直接继承 flask_login.UserMixin 类就可以了。

is_authenticated # 属性,默认返回 true
is_active # 属性,默认返回 true
is_anonymous # 属性,默认返回 False
get_id() # 方法,默认返回 id 属性的字符串表示

Flask-Login 内置了基于表单的一些辅助方法,在这里我们就不展开了。本文的开发方向是针对富客户端的应用,后端只提供 API。

除了直接利用 session 中的 user_id 来加载用户之外,还可以直接接管 request,从 request 中的 header 或者其他 token 来验证用户。

@manager.request_loader
def login_from_request(request):
    token = request.headers.get("X-Token")
    user = get_user_with_token(token)
    return user

如果当一个请求没有读取到用户,也就是用户是 None 的时候,Flask-Login 会使用内置的 AnoynmousUserMixin 来生成一个匿名用户。

is_authenticated # False
is_active # False
is_anonymous # True
get_id() # None

如果需要在用户未登录的时候,显示登录界面,或者返回 403 forbidden 等信息,应该使用 unauthorized_handler。

@manager.unauthorized_handler
def handle_login():
    return {"error": "need login"}

登录验证的 view

只需要添加 @login_required 就可以保护某个 view 需要登录了。

from flask_login import login_required

@login_required
def settings():
    pass

fresh login

在一些敏感的操作,比如需要改密码的时候,我们一般都要重新验证一次密码,这时候可以使用 fresh login 这个概念。

这里先不展开了。

flask admin

https://github.com/flask-admin/flask-admin/blob/master/examples/peewee/app.py

使用 swagger 生成文档

swagger 是一套定义 API 的工具,可以实现 API 的文档化和可交互。flasgger 是 flask 的一个插件,可以实现在注释中使用 swagger 语法。

swagger 本身是一套工具,但是后来被社区发展成了 OpenAPI 规范。最新版本是 OpenAPI 3.0,而现在用的最多的是 swagger 2.0。我们这里

完整的例子

https://github.com/coleifer/peewee/blob/master/examples/twitter/app.py

使用 uwsgi 部署

开发阶段使用的是 flask 内置的 debug server 来提供服务的。在生产环境部署的时候,我们则需要使用 uwsgi 这种多线程的服务器来提供更好的性能。

参考文献

  1. https://blog.csdn.net/u010466329/article/details/78522992
  2. https://blog.csdn.net/qq_21794823/article/details/78194164
  3. http://www.manongjc.com/article/48448.html
  4. https://juejin.im/post/5964ce816fb9a06bb21abb23
  5. https://www.cnblogs.com/whitewolf/p/4686154.html
  6. 为什么要使用 APP Factory Pattern
  7. https://flask-login.readthedocs.io/en/latest/

云时代的个人存储搭建

昨天想用 iPad 上的 GoodReader 看一本书,但是从 iCloud 同步的时候出了些问题,进度始终为零。由于国内糟糕的网络环境,这种同步失败的问题时有发生。虽然可以直接通过 WiFi 把书从电脑上传过来,但是因为偶尔需要在另一个 iPad 上查看,为了同步进度,还是最终决定还是自己搭建一套云存储设施。

ftp 与 webdav

ftp 协议有诸多问题,现在用的已经很少了。WebDav 协议基于 HTTP,相比 FTP 有不少有点,可以参见文章1。另外不少开源的网盘客户端也支持 webdav。NextCloud 支持 webdav,后面会讲到

sftp 和 sshfs

sftp 则和 ftp 是完全独立的两个东西,虽然最终目的是一样的。好比海豚和鲨鱼都是在海里的生物,但是一个是哺乳动物,而一个是鱼类。sftp 基于 ssh 协议。

sshfs 相比 sftp 则更近了一步,通过 sftp 把远程的文件系统直接映射到本地,从而无缝衔接。

搭建

sftp 直接基于 linux 的用户和文件权限系统。

  1. 添加相应的用户和分组,以用户名 sftp,分组名 ftpaccess 为例。
% sudo groupadd ftpaccess
% sudo useradd -m sftp -g ftpaccess -s /usr/sbin/nologin
% sudo passwd sftp  # 更改密码
% sudo mkdir /var/sftp
% sudo chown root /var/sftp  # 这一步非常坑,切记不可省略,后面讲为什么
% sudo mkdir -p /var/sftp/files
% sudo chown sftp:ftpaccess /var/sftp/files
  1. 修改 /etc/ssh/sshd_config 文件

注释掉这一行 Subsystem sftp /usr/lib/openssh/sftp-server

然后在文件的结尾添加

Subsystem sftp internal-sftp
Match group ftpaccess
ChrootDirectory /var/sftp  # 这里可以随便指定你想要的顶级目录
X11Forwarding no
AllowTcpForwarding no
ForceCommand internal-sftp
PasswordAuthentication yes

ssh 的安全配置要求 ChrootDirectory 本身必须是 root 所有的,所以登录都的根目录我们是不可写的,但是可以在新建的目录中读写。

  1. 重启 ssh 服务
% sudo systemctl restart ssh

可以使用客户端链接啦~ 如果需要使用 publickey 登录的话,只需要像普通的用户一样,把文件传到 ~sftp 的对应目录就可以了。

使用 sshfs mount 到本地

% brew install sshfs
% brew cask install osxfuse
% sshfs -o allow_other,defer_permissions -o volname=sftp_files sftp@your.example.com:/files $HOME/sftp_files

nextcloud

未完待续

参考资料

  1. https://stackoverflow.com/questions/11216884/which-file-access-is-the-best-webdav-or-ftp
  2. SSHFS
  3. 搭建 sftp 服务器
  4. sftp 的一个坑

Django 中使用多个数据库

有时候我们的表并不都在一个数据库中,需要使用多个数据库,django 支持配置并使用多个数据库。

定义多个数据库

首先,在 DATABASES 中定义需要使用的多个数据库:

DATABASES = {
    "default": {},
    "users": {
        "NAME": "user_data",
        "ENGINE": "django.db.backends.mysql",
        "USER": "mysql_user",
        "PASSWORD": "superS3cret"
    },
    "customers": {
        "NAME": "customer_data",
        "ENGINE": "django.db.backends.mysql",
        "USER": "mysql_cust",
        "PASSWORD": "veryPriv@ate"
    }
}

注意其中 default 是必须的,不过用不到的话,留空也行。

在使用 manage.py 的时候可以使用 --database=xxx 里指定数据库。

数据库路由

可以通过实现 Database Router 来让 django 自动选择应该使用的数据库。

DB router 需要实现下面四个方法,用来指定不同的 Model 对应的模型。

  1. db_for_read(model, **hints) 用来读取表时,查找对应的数据库。返回数据库配置名(DATABASES中定义的)
  2. db_for_write(model, **hints) 用来写入表时,查找对应的数据库。
  3. allow_relation
  4. allow_migrate

最后使用 DATABASE_ROUTERS 安装对应的路由:

DATABASE_ROUERS = ["path.to.router"]

django 单元测试

和普通的单元测试不同的是,django 单独提供了一个测试模块,所有的 TestCase 需要继承 django.test.TestCase

简单的测试

from django.test import TestCase
from myapp.models import Animal


class AnimalTestCase(TestCase):
    def setUp(self):
        Animal.objects.create(name="lion", sound="roar")
        Animal.objects.create(name="cat", sound="meow")

    def test_animals_can_speak(self):
        """Animals that can speak are correctly identified"""
        lion = Animal.objects.get(name="lion")
        cat = Animal.objects.get(name="cat")
        self.assertEqual(lion.speak(), "The lion says "roar"")
        self.assertEqual(cat.speak(), "The cat says "meow"")

对于需要测试服务器的测试用例,可以使用 django.test.Client

from django.test import TestCase

class SimpleTest(TestCase):
    def test_details(self):
        response = self.client.get("/customer/details/")
        self.assertEqual(response.status_code, 200)

    def test_index(self):
        response = self.client.get("/customer/index/")
        self.assertEqual(response.status_code, 200)

django 国际化

settings.py 中的设置:

MIDDLEWARE_CLASSES = (
    # ...
    "django.middleware.locale.LocaleMiddleware",
)


LANGUAGE_CODE = "en"
TIME_ZONE = "UTC"
USE_I18N = True
USE_L10N = True
USE_TZ = True

LANGUAGES = (
    ("en", ("English")),
    ("zh-hans", ("中文简体")),
    ("zh-hant", ("中文繁體")),
)

#翻译文件所在目录,需要手工创建
LOCALE_PATHS = (
    os.path.join(BASE_DIR, "locale"),
)

TEMPLATE_CONTEXT_PROCESSORS = (
    ...
    "django.core.context_processors.i18n",
)

生成需要翻译的文件:

python manage.py makemessages -l zh_hans
python manage.py makemessages -l zh_hant

翻译其中的 django.po 文件,注意.po文件是一种通用的格式,有很多专门的编辑器

编译翻译好的文件

python manage.py compilemessages

django 页面缓存

django 作为一个动态的网站系统,在并发访问量大的时候会遇到性能问题,这时候可以使用缓存来显著提高性能。

settings.py 中的配置

可以使用 django-redis 来使用 redis 作为缓存。

pip install django-redis
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
        }
    }
}

配置需要缓存的函数

from django.views.decorators.cache import cache_page

@cache_page(60 * 15) # 秒数
def index(request):
    # 读取数据库等 并渲染到网页
    return render(request, "index.html", {"queryset":queryset})

django 静态文件

settings.py 中的相关配置

STATIC_URL = "/static/"
STATIC_ROOT = os.path.join(BASE_DIR,"static")

一般来说我们只要把静态文件放在 APP 中的 static 目录下,部署时用 python manage.py collectstatic 就可以把静态文件收集到(复制到) STATICROOT 目录,但是有时我们有一些共用的静态文件,这时候可以设置 STATICFILESDIRS 另外弄一个文件夹,如下:

STATICFILES_DIRS = (
    os.path.join(BASE_DIR, "common_static"),
    "/var/www/static/",
)

这样我们就可以把静态文件放在 common_static 和 /var/www/static/中了,Django 也能找到它们。

MEDIA_URL = "/media/"
MEDIA_ROOT = os.path.join(BASE_DIR,"media")

media 文件夹用来存放用户上传的文件

nginx 部署时的配置

location /media  {
    alias /path/to/project/media;
}

location /static {
    alias /path/to/project/collected_static;
}

在模板中引入静态文件

{% load static %}
<img src="{% static "img/example.jpg" %}" alt="My image"/>

Django views 视图

django 使用正则指定路径,然后使用一个函数来处理对应的请求。

定义响应函数

响应函数如下:

# views.py

from django.shortcuts import render
from django.http import HttpResponse

def add(request):
    a = request.GET["a"]
    b = request.GET["b"]
    c = int(a) + int(b)
    return HttpResponse(str(c))


def add2(request, a, b):
    c = int(a) + int(b)
    return HttpResponse(str(c))

注意每个函数都需要接受 request 作为第一个参数,GET参数和POST参数都可以从 request 中读取。另外还可以使用从 url path 中读取数据,这些参数作为形参传递给对应的函数。

定义 URL 路由

# urls.py

from calc import views as calc_views

urlpatterns = [
    path("add/", calc_views.add, name="add"),  # new
    path("admin/", admin.site.urls),
    path("add/<int:a>/<int:b>/", calc_views.add2, name="add2"),  # django 2.0 的新语法,以前都是用正则分组
]

其中的 name 可以用在模板中,这样就不用写死 url 了。<a href="{% url 'add2' 4 5 %}">link</a>

设定响应的 headers

response 对象可以当做字典使用,向其中复制就可以设定响应的头部

from django.http import HttpResponse

def add(request):
    a = request.GET["a"]
    b = request.GET["b"]
    c = int(a) + int(b)
    response = HttpResponse(str(c))
    response["Powered-By"] = "django"
    return response

url reverse

在 urls.py 中可以设定 url 到具体函数的映射,但是 url 也是经常要随业务改动的,比如从 add/ 变成了 plus/。当我们在某个网页中需要链接到某个页面的时候,不希望写死 url,这时候可以使用 url reverse 的功能,使用 name 反向获取 url。

# urls.py
path("add/<int:a>/<int:b>/", calc_views.add2, name="add2")

# other.py
from django.urls import reverse
url = reverse("add2")

django forms

django 中的 form 和 model 的用法很像,都是定义一个类,然后指定一些字段就可以了

最简单的form

from django import forms

class ContactForm(forms.Form):
    subject = forms.CharField(max_length=100)
    email = forms.EmailField(required=False)
    message = forms.CharField(widget=forms.Textarea)

    def clean_message(self):
        message = self.cleaned_data['message']
        num_words = len(message.split())
        if num_words < 4:
            raise forms.ValidationError("Not enough words!")
        return message
def contact(request):
    if request.method == 'POST':
        form = ContactForm(request.POST)
        if form.is_valid():
            cd = form.cleaned_data
            send_mail(
                cd['subject'],
                cd['message'],
                cd.get('email', 'noreply@example.com'),
                ['siteowner@example.com'],
            )
            return HttpResponseRedirect('/contact/thanks/')
    else:
        form = ContactForm()
        return render(request, 'contact_form.html', {'form': form})
<form action="" method="post">
    <table>
        {{ form.as_table }}
    </table>
    {% csrf_token %}
    <input type="submit" value="Submit">
</form>

方法 | 用法
——|——
form.__str__() | return table representation
form.asp() | return p representation
form.as
li() | return li representation
form.__getitem__() | return element tag
form.__init__(dict) | fill values
form.isbound |
form.is
valid() |
form.cleaned_data |

Note not include table/ul/form tags, just the inside tags

ajax

ajax 中如何指定 crsf token

axios 中:

import axios from 'axios';
axios.defaults.xsrfHeaderName = "X-CSRFTOKEN";
axios.defaults.xsrfCookieName = "csrftoken";

settings.py 中

CSRF_COOKIE_NAME = "csrftoken"

参考

https://stackoverflow.com/questions/39254562/csrf-with-django-reactredux-using-axios

django dump to csv

import csv
from django.db.models.loading import get_model

def dump(qs, outfile_path):
    """
    Takes in a Django queryset and spits out a CSV file.

    Usage::

        >> from utils import dump2csv
        >> from dummy_app.models import *
        >> qs = DummyModel.objects.all()
        >> dump2csv.dump(qs, './data/dump.csv')

    Based on a snippet by zbyte64::

        http://www.djangosnippets.org/snippets/790/
    """
    model = qs.model
    writer = csv.writer(open(outfile_path, 'w'))

    headers = []
    for field in model._meta.fields:
        headers.append(field.name)
    writer.writerow(headers)

    for obj in qs:
        row = []
        for field in headers:
            val = getattr(obj, field)
            if callable(val):
                val = val()
            if type(val) == unicode:
                val = val.encode("utf-8")
            row.append(val)
        writer.writerow(row)