japronto備忘録(1)で、プロジェクトテンプレートを作成したが、Djangoのadmin機能のようなものがないと、開発効率が上がらないので、elixirのphoenixや、phpのphalconのdev-toolsようなCRUDのアプリを自動的に生成する仕組み(なんちゃってですが)を作成してみたのでそのアプリを投稿します。
1. 仕様
データベースの列情報、プライマリキー情報を取得して、以下の３つの処理を呼び出し、プロジェクトにテーブルのCRUDアプリを自動的に追加します。具体的には、以下の処理を行います。
(1). サービス層(project/apps/[テーブル名]/service.py)の作成
- 関数find_all()を出力する。
- 関数find_by_id()を出力する。
- 関数insertdata()を出力する。
- 関数updatedata()を出力する。
- 関数deletedata()を出力する。
(2). View層(project/apps/[テーブル名]/views.py)の作成
- 関数index()を出力する。
- 関数create()を出力する。
- 関数update()を出力する。
- 関数delete()を出力する。
- 上記を、japrontoのルータに追加する。
(3). HTMLテンプレートの作成
- project/templates/[テーブル名]/index.htmlを出力します。
- project/templates/[テーブル名]/create.htmlを出力します。
- project/templates/[テーブル名]/update.htmlを出力します。
- project/templates/[テーブル名]/delete.htmlを出力します。
※内容については、japronto備忘録(1)にサンプルファイルを掲載しています。
2. フォルダ／ファイル構成
│ createcrud.py
│
└─packages
│ create_services.py
│ create_views.py
│ create_templates.py
│ tableinfo.py
3. ソース
以下、ソースです。ツールなので、あまりきれいにかけてません、ご容赦ください。
(1) packages/create_services.py
import os
FIND_ALL_FUNC ="""
def find_all(cur):
sql = \"select * from {tablename}\"
dict_result = []
cur.execute(sql)
rows = cur.fetchall()
for row in rows:
dict_result.append(dict(row))
return dict_result
"""
def get_find_by_all(columndefs):
return FIND_ALL_FUNC.format(tablename=columndefs.tablename).strip()
FIND_BY_ID_FUNC ="""
def find_by_id(cur, query):
sql = \"select * from {tablename} {where} \"
cur.execute(sql, ({params}))
row = cur.fetchone()
return dict(row)
"""
def get_find_by_id(columndefs):
where = ""
joinstr =" where"
for pk in columndefs.pk_dict:
where+= f"{joinstr} {pk} = %s"
joinstr=" and"
params =""
for pk in columndefs.pk_dict:
params+= f"query['{pk}'],"
return FIND_BY_ID_FUNC.format(
tablename=columndefs.tablename, where=where, params=params ).strip()
INSERT_FUNC ="""
def insertdata(cur, query):
print(query)
sql = \"insert into {tablename} ({columns}) values ({values})\"
cnt = cur.execute(sql, ({params}))
return cnt
"""
def get_insert(columndefs):
columns = ""
params = ""
values = ""
comma =""
for column in columndefs.columns_dict:
#新規の場合、デフォルト値の設定がある場合は入力不要なのでinsertの項目を出力しない
if not (column['column_default']):
columns+=f"{comma}{column['column_name']}"
params+= f"query['{column['column_name']}'], "
values+=f"{comma}%s"
comma=","
return INSERT_FUNC.format(
tablename=columndefs.tablename,
columns=columns,
values=values,
params=params,
).strip()
UPDATE_FUNC ="""
def updatedata(cur, query):
print(query)
sql = \"update {tablename} set {columns} {where}\"
cnt = cur.execute(sql, ({params}))
return cnt
"""
def get_update(columndefs):
columns = ""
params = ""
comma = ""
for column in columndefs.columns_dict:
#変更の場合、制約の設定がある場合は入力不要なのでinsertの項目を出力しない
if not column['constraint_name']:
columns+=f"{comma}{column['column_name']} = %s"
params+= f"query['{column['column_name']}'], "
comma=","
where = ""
joinstr =" where"
#プライマリキーから、where句を出力する
for pk in columndefs.pk_dict:
where+= f"{joinstr} {pk} = %s"
joinstr=" and"
for pk in columndefs.pk_dict:
params+= f"query['{pk}'], "
return UPDATE_FUNC.format(
tablename=columndefs.tablename,
columns=columns,
where=where,
params=params,
).strip()
DELETE_FUNC ="""
def deletedata(cur, query):
print(query)
sql = \"delete from {tablename} {where}\"
cnt = cur.execute(sql, ({params}))
return cnt
"""
def get_delete(columndefs):
where = ""
joinstr =" where"
for pk in columndefs.pk_dict:
where+= f"{joinstr} {pk} = %s"
joinstr=" and"
params =""
for pk in columndefs.pk_dict:
params+= f"query['{pk}'], "
return DELETE_FUNC.format(
tablename=columndefs.tablename,
where=where,
params=params,
).strip()
def create_service(app_path, columndefs, methods, admin, schema):
dirname = os.path.join(app_path, columndefs.tablename)
if (not os.path.exists(dirname)):
os.mkdir(dirname)
filepath = os.path.join(dirname, "service.py")
with open(filepath, 'w')as f:
f.write(get_find_by_all(columndefs))
f.write("\n")
f.write("\n")
f.write(get_find_by_id(columndefs))
f.write("\n")
f.write("\n")
f.write(get_insert(columndefs))
f.write("\n")
f.write("\n")
f.write(get_update(columndefs))
f.write("\n")
f.write("\n")
f.write(get_delete(columndefs))
print(filepath)
def write_services(prjpath, columndefs, methods, admin, schema):
app_path = os.path.join(prjpath, "apps")
if (not os.path.exists(app_path)):
os.mkdir(app_path)
create_service(app_path, columndefs, methods, admin, schema)
(2) packages/create_views.py
import os
def write_header(f, columndefs):
f.write("from api.app import app, controller, get_tpl\n")
f.write("from psycopg2.extras import DictCursor\n")
f.write("from apps.%s.service import find_all, find_by_id, insertdata, updatedata, deletedata\n" % columndefs.tablename)
f.write("\n")
def write_comon(f, columndefs, crud, admin):
f.write("@controller\n")
f.write("def %s(request):\n" % crud)
f.write("\tloginuser = request.is_loggedin\n")
if admin:
f.write("\thtml=get_tpl('/admin/%s/%s.html')\n" % (columndefs.tablename, crud))
else:
f.write("\thtml=get_tpl('/%s/%s.html')\n" % (columndefs.tablename, crud))
f.write("\tresult={}\n")
def write_index(f, columndefs, crud):
f.write("\tconn = request.connection\n")
f.write("\twith conn.cursor(cursor_factory=DictCursor) as cur:\n")
f.write("\t\tresult=find_all(cur)\n")
def write_create(f, columndefs, crud):
f.write("\tif request.method.lower()==\"post\":\n")
f.write("\t\tconn = request.connection\n")
f.write("\t\twith conn.cursor(cursor_factory=DictCursor) as cur:\n")
f.write("\t\t\tresult=insertdata(cur, request.form)\n")
f.write("\t\tconn.commit()\n")
def write_update(f, columndefs, crud):
f.write("\tconn = request.connection\n")
f.write("\twith conn.cursor(cursor_factory=DictCursor) as cur:\n")
f.write("\t\tif not request.method.lower()==\"post\":\n")
f.write("\t\t\tresult=find_by_id(cur, request.query)\n")
f.write("\t\telse:\n")
f.write("\t\t\tresult=updatedata(cur, request.form)\n")
f.write("\t\tconn.commit()\n")
def write_delete(f, columndefs, crud):
f.write("\tconn = request.connection\n")
f.write("\twith conn.cursor(cursor_factory=DictCursor) as cur:\n")
f.write("\t\tif not request.method.lower()==\"post\":\n")
f.write("\t\t\tresult=find_by_id(cur, request.query)\n")
f.write("\t\telse:\n")
f.write("\t\t\tresult=deletedata(cur, request.form)\n")
f.write("\t\tconn.commit()\n")
def create_views(app_path, columndefs, methods, admin):
filepath=os.path.join(app_path, columndefs.tablename+"/views.py")
with open(filepath, 'w')as f:
write_header(f, columndefs)
for crud in methods:
write_comon(f, columndefs, crud, admin)
if crud=="index":
write_index(f, columndefs, crud)
elif crud=="create":
write_create(f, columndefs, crud)
elif crud=="update":
write_update(f, columndefs, crud)
elif crud=="delete":
write_delete(f, columndefs, crud)
f.write("\treturn request.Response(text=html.render(data=result, user=loginuser),mime_type='text/html') \n")
f.write("\n")
for crud in methods:
url=""
if admin:
url="/admin"
url+="/"+columndefs.tablename
if not crud=="index":
url+="/"+crud
f.write("app.router.add_route('%s', %s)\n" % (url, crud))
print(filepath)
def write_views(prjpath, columndefs, methods, admin):
app_path = os.path.join(prjpath, "apps")
if (not os.path.exists(app_path)):
os.mkdir(app_path)
create_views(app_path, columndefs, methods, admin)
(3) packages/create_templates.py
import os
_btnsubmit="<button class='btn btn-sm btn-success ml-2' type='submit'>submit</button>"
def get_title(crud, tablename):
return "<h1>%s %s</h1>\n" % (tablename, crud, )
def get_url_path(tablename, crud, admin):
urladmin = "/admin" if admin else ""
url = "/"+tablename if crud=="index" else "/"+tablename+"/"+crud
return "%s%s" %(urladmin, url)
def get_anchor(tablename, crud, admin, params=None):
url = get_url_path(tablename, crud, admin)
if params:
url+=params
btnclass="btn-primary"
if crud=="update":
btnclass="btn-info"
elif crud=="delete":
btnclass="btn-danger"
return "<a class='btn btn-sm %s ml-2' style='width: 60px;' href='%s'>%s</a>" %(btnclass, url, crud)
def output_index(f,columndefs, admin):
get_params=""
joinstr="?"
f.write("{% extends 'base.html' %}\n{% block content %}\n")
f.write(get_title("list", columndefs.tablename))
f.write("<table class='table table-sm table-striped table-borderd'>\n")
f.write("<thead>\n<tr>\n")
for row in columndefs.columns_dict:
f.write("<th>%s</th>" % row["column_name"])
if row["column_name"] in columndefs.pk_dict:
get_params+="%s%s={{ row.%s }}" % (joinstr, row["column_name"], row["column_name"])
joinstr="&"
anchor = get_anchor(columndefs.tablename, "create", admin)
f.write("<th style='width: 160px;'>%s</th>\n" % (anchor, ))
f.write("</tr>\n</thead>\n")
f.write("<tbody>\n")
f.write("{% for row in data %}\n")
f.write("<tr>\n")
for row in columndefs.columns_dict:
f.write("")
f.write("<td>{{ row.%s }}</td>\n" % row["column_name"])
updateanchor = get_anchor(columndefs.tablename, "update", admin, get_params)
deleteanchor = get_anchor(columndefs.tablename, "delete", admin, get_params)
f.write("<td>%s%s</td>\n" % (updateanchor, deleteanchor ))
f.write("</tr>\n")
f.write("{% endfor %}\n")
f.write("</tbody>\n</table>\n")
f.write("{% endblock %}\n")
def output_cud(f,columndefs, method, admin):
f.write("{% extends 'base.html' %}\n{% block content %}\n")
f.write(get_title(method, columndefs.tablename))
url=get_url_path(columndefs.tablename, method, admin)
f.write("\t<form class='needs-validation' method='post' action='%s' enctype='application/x-www-form-urlencoded'>\n" % (url,))
require=""
readonly=""
style=""
inputtype="text"
if method=="delete":
readonly="readonly"
for row in columndefs.columns_dict:
# print("%s is_null=%s" % (row['column_name'], row['is_nullable']))
if (method in ("create", "update") and row['is_nullable']=='NO'):
require='required'
# print("require=%s" % require)
if not (method=="create" and row['column_default']):
if row['column_default']:
style="style='display:none;'"
else:
style=""
f.write("\t\t<div class='form-group' %s>\n" % (style, ))
f.write("\t\t\t<label class='form-label'>%s</label>\n" % (row["column_name"]))
if row["column_name"]=="password":
inputtype="password"
else:
inputtype="text"
if method in ("update", "delete", ):
# value="value='{{ data.%s }}'"
f.write("\t\t\t<input class='form-control' type='%s' name='%s' value='{{ data.%s }}' %s %s/>\n" %
(inputtype, row["column_name"], row["column_name"], readonly, require))
else:
f.write("\t\t\t<input class='form-control' type='%s' name='%s' %s/>\n" %
(inputtype, row["column_name"], require, ))
f.write("\t\t</div>\n")
gobackbtn = get_anchor(columndefs.tablename, "index", admin)
f.write("\t\t%s\n%s\n" % (_btnsubmit, gobackbtn))
f.write("\t</form>\n")
f.write("{% endblock %}\n")
def write_templates(prjpath, columndefs, methods, admin):
templates_root_path = os.path.join(prjpath, "templates")
if admin:
adminroot = os.path.join(templates_root_path, "admin")
templates_root_path = adminroot
if (not os.path.exists(templates_root_path)):
os.mkdir(templates_root_path)
templates_path = os.path.join(templates_root_path, "%s" % columndefs.tablename)
if (not os.path.exists(templates_path)):
os.mkdir(templates_path)
for method in methods:
filepath = os.path.join(templates_path, method + ".html")
with open(filepath, 'w') as f:
if method == "index":
output_index(f, columndefs, admin)
else:
output_cud(f, columndefs, method, admin)
print("create template path=%s" % filepath)
(4) packages/tableinfo.py
プロジェクトのデータベース情報を読み込んでデータベースに接続し、テーブルの列およびプライマリキー情報を取得します。
import psycopg2
import psycopg2.extras
import configparser
_sql = "select " \
" col.column_name " \
" ,col.ordinal_position " \
" ,col.column_default " \
" ,col.is_nullable " \
" ,col.data_type " \
" ,col.character_maximum_length as max_length " \
" ,col.character_octet_length as oct_length " \
" ,col.numeric_precision as num_precision " \
" ,col.numeric_precision_radix as num_radix " \
" ,col.numeric_scale as num_scale " \
" ,col.datetime_precision as dt_precision " \
" ,cu.constraint_name " \
" ,kcu.ordinal_position as pk_ordinal " \
"from " \
" information_schema.columns col " \
"left join " \
" information_schema.constraint_column_usage cu " \
"on " \
" col.table_catalog=cu.table_catalog " \
"and " \
" col.table_schema=cu.table_schema " \
"and " \
" col.table_name=cu.table_name " \
"and " \
" col.column_name=cu.column_name " \
"left join " \
" information_schema.key_column_usage kcu " \
"on " \
" cu.constraint_catalog = kcu.constraint_catalog " \
"and " \
" cu.constraint_schema = kcu.constraint_schema " \
"and " \
" cu.constraint_name = kcu.constraint_name " \
"and " \
" cu.table_catalog = kcu.table_catalog " \
"and " \
" cu.table_schema = kcu.table_schema " \
"and " \
" cu.table_name = kcu.table_name " \
"and " \
" cu.column_name = kcu.column_name " \
"where " \
" col.table_catalog=%s " \
"and " \
" col.table_schema=%s " \
"and " \
" col.table_name=%s " \
"order by " \
" col.ordinal_position "
class ColumnDefs:
def __init__(self, prjpath, tablename, schema):
config = configparser.ConfigParser()
config.read(prjpath+"/config/database.conf")
host=config['DB']['host']
port=config['DB']['port']
dbname=config['DB']['dbname']
user=config['DB']['user']
password=config['DB']['password']
dsn = f"host={host} port={port} dbname={dbname} user={user} password={password}"
#
self.tablename = tablename
self.pk_dict = list()
self.columns_dict = []
with psycopg2.connect(dsn) as conn:
with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
schemaname = "public" if not schema else schema
cur.execute(_sql, (dbname, schemaname, tablename))
rows = cur.fetchall()
if len(rows)==0:
print("Error ColumnDef Not Found!!")
print(_sql)
print((dbname, schemaname, tablename))
for row in rows:
dictrow = dict(row)
# print(dictrow)
if dictrow['pk_ordinal'] is not None:
self.pk_dict.append(dictrow['column_name'])
self.columns_dict.append(dictrow)
# print(self.columns_dict)
(5) createcrud.py
実行用。プロジェクトのパス、テーブル名などを引数に処理を実行する、メイン処理です。
import os
import sys
import argparse
from packages.tableinfo import ColumnDefs
from packages.create_services import write_services
from packages.create_views import write_views
from packages.create_templates import write_templates
parser = argparse.ArgumentParser()
parser.add_argument("projectpath", type=str, help="need <project_abs_path>")
parser.add_argument("table", type=str, help="need tablename")
parser.add_argument("--schema", type=str, help="--schema=<schemaname>")
parser.add_argument("--admin", help="--admin", action="store_true")
args = parser.parse_args()
try:
if args.schema:
print("schema=%s" % args.schema)
if args.admin:
print("admin controller")
PROJECT_PATH = args.projectpath
TABLE_NAME = args.table
print("tablename="+TABLE_NAME)
columndefs = ColumnDefs(PROJECT_PATH, TABLE_NAME, args.schema )
if len(columndefs.columns_dict)==0:
errmsg = """Error!! not found table.
confirm your database. table={0} schema={1}
""".format(TABLE_NAME,"public" if not args.schema else args.schema)
raise TableNotFoundException( errmsg )
METHODS = ["index","create","update","delete"]
write_services(PROJECT_PATH, columndefs, METHODS, args.admin, args.schema)
write_views(PROJECT_PATH, columndefs, METHODS, args.admin)
write_templates(PROJECT_PATH, columndefs, METHODS, args.admin)
except:
print("Error!! not found table. confirm your database. table=%s" % TABLE_NAME)
traceback.print_exc()
sys.exit()
4. 使い方
(1). ターゲットとなるデータベース(postgreSQL)にテーブルを追加します。
create table "public".todos (
id bigserial primary key
, todo character varying(255) not null
);
(2). ターゲットとなるプロジェクトのデータベース(postgreSQL)情報を設定します。
conf/database.conf
[DB]
minconn=10
maxconn=50
host=localhost
port=5432
dbname=japrontodb
user=testuser
password=************************
(3). コマンドプロンプトから、以下のコマンドを実行します。
python .\createcrud.py C:\linux_home\japronto\mysite todos
(4). プロジェクトファイルに追加したアプリを設定する。
templates/base.htmlの「header」ブロックに、追加したアプリのURLを追加します。
templates/base.html
・・・・（省略）・・・・・
<header style="background: #263238;padding-left:5px;padding-right:5px;">
<div class="navi">
<div class="navi left" style="width:70%; ">
<div class="title">My Project</div>
{% if user %}
<div>
<a href="/todos">todo</a>
</div>
{% endif %}
</div>
・・・・（省略）・・・・・
5. 出来上がり
python main.py
