Python+ResponderでWEBアプリケーションを構築する(2)

Python+ResponderでWEBアプリケーションを構築するの続きです。

※随時更新します

(1) 棒グラフを作成する。

matplotlib, numpyをインストールします。

cmd.prompt

pip install matplotlib
pip install numpy

main.py

import responder
import io
import matplotlib.pyplot as plt
import numpy as np

api = responder.API()

# create the plot
def createPlotAndSvg():
    #棒グラフを作成する。
    x = np.arange(3)
    y = np.array([100, 30, 70])
    plt.bar(x, y)

    #plotをSVGに変換する。
    buf = io.BytesIO()
    plt.savefig(buf, format='svg', bbox_inches='tight')
    # plotをクリア
    plt.cla()   
    svg = buf.getvalue()
    buf.close()
    return svg

@api.route("/")
async def bargraph(req, resp):
    svg = createPlotAndSvg()
    resp.headers['content_type'] = 'image/svg+xml'
    resp.content = svg

api.run()

(2) markdown

markdownをインストールします。

cmd.prompt

pip install markdown

main.py

import responder
import markdown

api = responder.API()

@api.route("/")
async def hello(req, resp):
    md_text="" \
    "# 見出し 1\n" \
    "## 見出し 2\n" \
    "### 見出し 3\n" \
    "**強調**\n" \
    "\n[リンク](http://...)\n\n" \
    "- リスト 1\n" \
    "- リスト 2\n" \
    "    - リスト 2-1\n" \
    ""
    html = markdown.markdown(md_text, safe_mode='escape')
    resp.headers = {"Content-Type": "text/html; charset=utf-8"}
    resp.text = html

api.run()

(3) プロジェクトテンプレート

responderを使ったプロジェクトのテンプレートを検討してみた。

テンプレートを使ったファイル/フォルダ構成

mysite
│  main.py
│  
├─apps
│     app.py
│     commons.py
│     db.py
│     dbproperties.py
│     properties.py
│          
├─settings
│      app.conf
│      logging.conf
│      
├─templates
│  │  layout.html
│  │  
│  │  welcome
│  │       welcome.html
│  └─sapleapp
│          sapleapp.html

以下、個別のプリケーションを追加していけるようにした。

│          
├─welcome
│      views.py
│
└─users
      models.py
      service.py
      views.py

static
├─css
│
└─js

テンプレートの簡単な説明

directory	file	説明
(root)	main.py	ルーティングを定義
apps	app.py	Responder、ログなどの生成
	commons.py	プロジェクトの共通クラスや関数を定義
	db.py	データベースのオブジェクト生成とDB用のfunctionを定義
	dbproperties.py	DB接続用のプロパティを管理
	properties.py	configファイルを読んで、アプリケーションの設定を管理
settings	app.conf	プロジェクトの各種プロパティを定義
	logging.conf	loggingのconfig

テンプレートの個々のソースファイル

• main.py

main.py

from apps.app import api

from welcome.views import WelcomeView
from users.views import UsersView, UsersSearchView

api.add_route("/", WelcomeView)
api.add_route("/users/findall", UsersSearchView)
api.add_route("/users/findbyid/{id}", UsersView)

api.run()

• apps/app.py

apps/app.py

# import os
import logging
from logging import config, getLogger
import yaml
import responder
from pathlib import Path

def create_app():
    # static dir
    ROOT_DIR = Path(__file__).resolve().parents[2]

    # logging setup
    config.dictConfig(
        yaml.load(open('settings/logging.conf', encoding='UTF-8').read(), Loader=yaml.FullLoader))
    logger = getLogger("mysite")

    # declare responder
    api = responder.API(
        static_dir=str(ROOT_DIR.joinpath('static')),
    )

    return api

api=create_app()

• apps/commons.py

apps/commons.py

from apps.db import Session

"""
dbのセッションを管理するクラスで、子オブジェクトにセッションの管理を意識させないようにしています
"""
class ServiceBase:
    def __init__(self):
        self.session = Session()
    print("start session")

    def __del__(self):
        self.session.close()
        print("close session")

• apps/db.py

apps/db.py

import traceback
import logging
import json
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from datetime import date, datetime
from decimal import Decimal
from sqlalchemy.orm.state import InstanceState
from apps.dbproperties import DBProp

prop = DBProp()

ECHO_LOG = False

engine = create_engine(
    prop.url, 
    echo=ECHO_LOG, 
    pool_size=prop.pool_size, 
    max_overflow=prop.max_overflow,
    isolation_level=prop.isolation_level,
)

Base = declarative_base()
Session = scoped_session(sessionmaker(bind=engine))

logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

'''
SQLAlchemy Modelまたはクエリの実行結果の行オブジェクト（またはその配列）をdict（の配列）に変換
'''
def row_to_dict(rowobj):
    if '__dict__' in dir(rowobj):
        return rowobj.__dict__
    else:
        return dict(rowobj)

'''
SQLAlchemyのormオブジェクトまたはsqlの実行結果（またはその配列）をdict（の配列）に変換
'''
def to_dict(obj):  #obj = sqlalchemy or cursor objects
    if isinstance(obj, list):
        for r in obj:
            return [row_to_dict(r) for r in obj]
    else:
        # Cursor resultset Rows
        return row_to_dict(obj)

'''
json変換のデフォルトを定義する。
'''
def json_converter(obj):
    if isinstance(obj, datetime):
        # 日時型は、文字列(isoformat(' '))に変換
        return obj.isoformat(' ', timespec='seconds')
    elif isinstance(obj, date):
        # 日付型は、文字列(isoformat)に変換
        # 【注意！】isinstanceは、datetime型なら、dateでもtrueになるので、
        # datetimeを先に判定してからdateを判定する。
        return obj.isoformat()
    elif isinstance(obj, Decimal):
        # Decimal型は、floatに変換
        return float(obj)
    elif isinstance(obj, InstanceState):
        # SQLAlchemyのオブジェクトのInstanceStateプロパティは変換しない。
        # 
        return ''
    raise TypeError ("Type %s not serializable" % type(obj))

'''
dictオブジェクト（またはその配列）をjsonに変換する。
'''
def to_json(objects):
    objdict = to_dict(objects)
    return json.dumps(objdict, default=json_converter, ensure_ascii=False)

• apps/dbproperties.py

apps/dbproperties.py

from apps.properties import props

class DBProp:
    def __init__(self):
        """コンストラクタ"""
        dialect = props.get('db', 'dialect')
        driver = props.get('db', 'driver')
        username = props.get('db', 'username')
        password = props.get('db', 'password')
        host = props.get('db', 'host')
        port = props.get('db', 'port')
        database = props.get('db', 'database')
        self._url = f"{dialect}+{driver}://{username}:{password}@{host}:{port}/{database}"
        self._pool_size = props.getint('db', 'pool_size')
        self._max_overflow = props.getint('db', 'max_overflow')
        self._isolation_level = props.get('db', 'isolation_level')

    @property
    def url(self):
        return self._url

    @property
    def pool_size(self):
        return self._pool_size

    @property
    def max_overflow(self):
        return self._max_overflow

    @property
    def isolation_level(self):
        return self._isolation_level

    @property
    def pool_size(self):
        return self._pool_size

• apps/properties.py

apps/properties.py

import configparser

def create_prop():
    conf = configparser.ConfigParser()
    conf.read('settings/app.conf')
    return conf

props = create_prop()

• settings/app.conf

settings/app.conf

[db]
dialect = mysql
driver = pymysql
username = responderuser
password = ************************
host = localhost
port = 3306
database = testdb
pool_size = 20
max_overflow = 0
isolation_level = READ UNCOMMITTED
・・・・・

• settings/logging.conf

settings/logging.conf

version: 1

formatters:
  custmoFormatter:
    format: '%(asctime)s %(levelname)s - %(filename)s %(funcName)s %(lineno)d: %(message)s'
    ; datefmt: '%Y/%m/%d %H:%M:%S'

loggers:
  test:
    handlers: [fileRotatingHandler,consoleHandler]
    level: DEBUG
    qualname: test
    propagate: no

  file:
    handlers: [fileRotatingHandler]
    level: DEBUG
    qualname: file
    propagate: no

  console:
    handlers: [consoleHandler]
    level: DEBUG
    qualname: console
    propagate: no

handlers:
  fileRotatingHandler:
    formatter: custmoFormatter
    class: logging.handlers.TimedRotatingFileHandler
    level: DEBUG
    filename: c:/mysite/log/application.log
    encoding: utf8
    when: 'D'
    interval: 1
    backupCount: 14

  consoleHandler:
    class: logging.StreamHandler
    level: DEBUG
    formatter: custmoFormatter
    stream: ext://sys.stdout

root:
  level: DEBUG
  handlers: [fileRotatingHandler,consoleHandler]

アプリケーションの実装（usersの例）

テンプレートを使うことで、Controller(views.py)を簡単に作成できます。また、サービス(service.py)をControllerと分離して、役割を明確化できます。

• users/models.py

users/models.py

from apps.db import Base
from sqlalchemy import Column, Integer, String, Text, text, DATETIME, DATE
from datetime import datetime
import json

class User(Base):
    __tablename__ = "users"
    id = Column('id', Integer, primary_key = True, autoincrement=True)
    username = Column('username', String(32))
    mailaddress = Column('mailaddress', String(255))
    password = Column('password', String(255))
    birthday = Column('birthday', DATE)
    role = Column('role', String(255))
    created_at = Column('created_at', DATETIME, nullable=False, default=datetime.now)
    updated_at = Column('updated_at', DATETIME, nullable=False, default=datetime.now, onupdate=datetime.now)

• users/service.py

アプリケーションに必要な処理は、すべてここに実装します。

users/service.py

from apps.commons import ServiceBase
from apps.db import to_dict
from users.models import User

class UserService(ServiceBase):
    def find_all(self):
        ret = self.session.execute("select * from users").fetchall()
        return to_dict(ret)

    def find_by_id(self, id):
        users = self.session.query(User).\
            filter(User.id==id).\
            first()
        return  to_dict(users)

• users/views.py

Controllerクラス。ResponderのClass-Based Viewsを利用しています。

users/views.py

from apps.app import api
from users.service import UserService
from apps.db import to_json

class UsersBaseView:
    def __init__(self):
        self.service = UserService()

    def __del__(self):
        del self.service

class UsersSearchView(UsersBaseView):
    async def on_request(self, req, resp):
        users = self.service.find_all()
        resp.content = api.template('users/users.html', list=users) 

class UsersView(UsersBaseView):
    async def on_request(self, req, resp, id=None):
        user = self.service.find_by_id(id)
        resp.content = api.template('users/user.html', data=user)