POSTED BY
2024-07-17

Python3.10でWSGI準拠の簡易HTTPサーバーを書いて常駐させる(静的ファイル対応)

続きです。前回作成した単独簡易WSGI-HTTPサーバーを拡張して、リクエストパラメータをGET/POSTそれぞれで取得してそのまま返すプログラムにします。

ソースコードはこちら。

import os
import urllib
from wsgiref.simple_server import make_server

CONTENT_TYPE = {
  '.html': 'text/html; charset=utf-8', '.txt': 'text/plain; charset=utf-8', '.json': 'application/json',
  '.js': 'text/javascript', '.css':'text/css', '.ico': 'image/vnd.microsoft.icon', '.jpg': 'image/jpeg',
  '.png': 'image/png', '.gif': 'image/gif', '.webp': 'image/webp'
};

def application(environ, start_response):
  content_type = 'text/plain; charset=utf-8'
  output_text = ''
  output_data = None
  path_info = environ['PATH_INFO']
  method = environ.get('REQUEST_METHOD')

  if 'get-test' in path_info:
    output_text = 'called get-test function.'
    if method == 'GET':
      wsgi_input = environ.get('QUERY_STRING')
      if len(wsgi_input) > 0:
        query = urllib.parse.parse_qs(wsgi_input)
        print(query)
        if 'input_text' in query:
          output_text = query['input_text'][0]
          print(output_text)
  elif 'post-test' in path_info:
    output_text = 'called post-test function.'
    if method == 'POST':
      content_length = int(environ.get('CONTENT_LENGTH', 0))
      if content_length > 0:
        wsgi_input = environ['wsgi.input'].read(content_length).decode('utf-8')
        query = urllib.parse.parse_qs(wsgi_input)
        print(query)
        if 'input_text' in query:
          output_text = query['input_text'][0]
          print(output_text)
  elif '.' in path_info:
    full_path = '.' + ( path_info + 'index.html' if path_info.endswith('/') else path_info)
    file_name, file_ext = os.path.splitext(full_path)
    if file_ext in CONTENT_TYPE:
      if os.path.isfile(full_path):
        with open(full_path, mode='br') as f:
          output_data = f.read()
          content_type = CONTENT_TYPE[file_ext]
      else:
        output_text = 'file path not found'
    else:
      output_text = 'not supported ext type'
  else:
    output_text = 'not supported file path'

  headers = [('Content-type', content_type)]
  start_response('200 OK', headers)

  if output_data != None:
    return [bytes(output_data)]

  return [bytes(output_text.encode('utf-8'))]

with make_server('', 8000, application) as httpd:
  print('make_server port 8000...')
  httpd.serve_forever()

・ブラウザ要求URIに「get-test」が含まれていたら、GETメソッドテストと見なし、QUERY_STRING文字列をurllib.parse.parse_qsにかけて連想配列に変換しています。また「input_text」というパラメータ名ならば、その値をそのままブラウザに返します。それ以外なら「called get-test function.」と返します。

・ブラウザ要求URIに「post-test」が含まれていたら、POSTメソッドテストと見なし、標準入力からCONTENT_LENGTHバイトをバイナリデータとして読み込みます。バイナリなので、文字列としてパースする前にdecode(utf-8)として変換する必要があります。あとはGETと同様、urllib.parse.parse_qsにかけて連想配列に変換しています。また「input_text」というパラメータ名ならば、その値をそのままブラウザに返します。それ以外なら「called get-test function.」と返します。

なおparse_qsの返す連想配列は１つのキーが複数の値を配列で持つ前提で返してくるので、基本的には値の取得は配列の先頭[0]を参照することになります。

バックグラウンド実行

python3 -m wsgi_server2.py > wsgi_server.log 2>&1 &

とすると、8000番ポートでHTTP待ち受けをバックグラウンドで開始する。
標準エラー出力を標準出力にまとめてリダイレクトしている。
出力はすべてwsgi_server.logに記録される。

テスト

GETのテスト

curl -X GET "http://127.0.0.1:8000/get-test?input_text=TEST"
TEST
curl -X GET "http://127.0.0.1:8000/get-test?test_text=TEST"
called get-test function.

POSTのテスト

curl -X POST "http://127.0.0.1:8000/post-test" -d "input_text=てすと"
てすと
curl -X POST "http://127.0.0.1:8000/post-test" -d "test_text=てすと"
called post-test function.