Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。Python 由 Guido van Rossum 于1989年底发明,第一个公开发行版发行于1991年。像Perl语言一样, Python 源代码同样遵循 GPL(GNU General Public License)协议。

最近使用Python语言实现了一个简单的App服务器,下面这两个文件中的源码Demo供大家参考。

“http_server.py” 实现了 Http 服务器的功能,其中 “ThreadingSimpleServer” 实现了多线程功能。

# http_server.py
from http.server import BaseHTTPRequestHandler, HTTPServer
from socketserver import ThreadingMixIn
from urllib.parse import urlparse
from v1_posts import api_v1

PORT_NUMBER = 8080

class ThreadingSimpleServer(ThreadingMixIn, HTTPServer):
    """Handle requests in a separate thread."""

# This class will handles any incoming request from
# the browser 
class HTTPRequestHandler(BaseHTTPRequestHandler):
    
    def ping_cb(self):
        self.send_response(200, 'ok')
        self.end_headers()
        # Send the html message
        self.wfile.write(b'PONG')
        
    def app_process_request(self):
        print('client is:', self.client_address)
        #print(self.command)
        #print(self.path)
        uri = urlparse(self.path)
        print(uri)
        
        if uri.path.lower() == '/ping':
            self.ping_cb()
        elif uri.path.lower() == '/posts/public_timeline':
            api_v1.posts_public_timeline(self)
        elif uri.path.lower() == '/news':
            api_v1.news(self)
        else:
            self.send_response(404)
            #self.send_header('Content-type','application/json;charset=UTF-8')
            self.end_headers()            
            # Send the html message            
            self.wfile.write(str.encode(str(404)))
            self.wfile.write(str.encode('\r\nNot Found'))
        return
        
    # Handler for the GET requests
    def do_GET(self):
        self.app_process_request()
        
    # Handler for the POST requests
    def do_POST(self):
        self.app_process_request()
    

try:
    # Create a web server and define the handler to manage the
    # incoming request
    server_address = ('', PORT_NUMBER);
    server = ThreadingSimpleServer(server_address, HTTPRequestHandler)
    print('Started httpserver on' , server.server_address, ', use <Ctrl-C> to stop')
    
    #Wait forever for incoming htto requests
    server.serve_forever()

except KeyboardInterrupt:
    print('^C received, shutting down the web server')
    server.socket.close()

“v1_posts.py” 实现了一个接口,其中有使用 MySQL 数据库和返回 JSON 数据的代码。

# v1_posts.py
# -*- coding:utf-8 -*- 
 
from urllib.parse import urlparse, parse_qsl
import time
import pymysql
import json

MYSQL_HOST = '192.168.2.20'
MYSQL_PORT = 3306
MYSQL_USER = 'root'
MYSQL_PASSWORD = 'password'
MYSQL_DB = 'test'

class api_v1:
    def posts_public_timeline(o):    
        postArrayObject = []
        uri = urlparse(o.path)
        querys = dict(parse_qsl(uri.query))
        print(querys)
        
        temp = querys.get('since_time', round(time.time() * 1000))
        try:
            since_time = int(temp)
        except:
            since_time = round(time.time() * 1000)
        #print(since_time)
        count = 20
            
        # 打开数据库连接
        is_db_open = False
        try:
            db = pymysql.connect(host=MYSQL_HOST, port=MYSQL_PORT, user=MYSQL_USER, password=MYSQL_PASSWORD, db=MYSQL_DB, charset='utf8mb4')
            is_db_open = True
            
            with db.cursor() as cursor:
                sql = 'SELECT * FROM `event_review` WHERE `area_code` = %s LIMIT 0,%s'
                cursor.execute(sql, ('451', count))
                result = cursor.fetchall()
                for row in result:                    
                    id = row[0]
                    uid = row[1]
                    posttime = round(row[2].timestamp() * 1000)
                    text = row[4]
                    
                    postObject = {}
                    postObject['id'] = id                    
                    postObject['posttime'] = posttime
                    postObject['text'] = text
                    
                    userObject = {}
                    userObject['id'] = uid
                    postObject['user'] = userObject
                    
                    postArrayObject.append(postObject)
                    
                    #print(id, uid, posttime, text)
        except pymysql.err.OperationalError as err:
            print(err)
        except pymysql.err.Error as err:
            print(err)
        except pymysql.err.Warning as err:
            print(err)    
        except Exception as err:
            print(err)
        except:
            print('unknown error!')
        
        # 关闭数据库连接
        if is_db_open:
            try:
                db.close();
            except pymysql.err.Error as err:
                print(err)
            except:
                print('unknown error!')
        
        jsonObject = {}
        jsonObject['error_code'] = 0
        jsonObject['error_msg'] = 'ok'
        jsonObject['results'] = postArrayObject
        
        # response
        o.send_response(200)
        o.send_header('Content-type','application/json;charset=UTF-8')
        o.end_headers()            
        # Send the josn message
        o.wfile.write(str.encode(json.dumps(jsonObject)))
        #print(json.dumps(jsonObject, indent=4))
        
        #print('public_timeline')

标签: http, Python, App服务器

添加新评论