Tornado

1.简介

要在 Tornado 中使用 HighGoDB ，你需要安装一个 Python 的 HighGoDB 驱动，通常使用 psycopg2 。

本文主要介绍 Tornado 该如何连接使用瀚高数据库。

2.加载psycopg2

参考 Psycopg2 接口文档。

3.连接瀚高CURD

tornado_demo.py

# -*- coding: utf-8 -*-
import tornado.ioloop
import tornado.web
import psycopg2
import json
 
# 连接数据库配置
DATABASE_CONFIG = {
    "host": "192.168.100.101",
    "dbname": "highgo",
    "user": "sysdba",
    "password": "Qwer@1234",
    "port": "5866",
}
 
# 连接数据库
def get_db_connection():
    return psycopg2.connect(**DATABASE_CONFIG)
 
class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("Database curd begin ... \n")
        # 创建表
        self.create_table()
 
        # 插入数据
        self.insert_data()
 
        # 查询数据
        res =  self.select_data()
        self.write(json.dumps(res))
        self.write("\n")
        
        # 更新数据
        self.update_data()
 
        # 删除数据
        self.delete_data() 
        
        self.write("Database curd end ... \n")
    
    # 创建表
    def create_table(self):
        conn = get_db_connection()
        cur = conn.cursor()
        cur.execute("CREATE TABLE IF NOT EXISTS test (id SERIAL PRIMARY KEY, name VARCHAR(100))")
        conn.commit()
        cur.close()
        conn.close()
 
    # 插入数据
    def insert_data(self):
        conn = get_db_connection()
        cur = conn.cursor()
        cur.execute("INSERT INTO test (name) VALUES ('test_name')")
        conn.commit()
        cur.close()
        conn.close()
 
    # 查询数据
    def select_data(self):
        conn = get_db_connection()
        cur = conn.cursor()
        cur.execute("SELECT * FROM test")
        data = cur.fetchall()
        cur.close()
        conn.close()
        return data
 
    # 更新数据
    def update_data(self):
        conn = get_db_connection()
        cur = conn.cursor()
        cur.execute("UPDATE test SET name = 'updated_name' WHERE id = 1")
        conn.commit()
        cur.close()
        conn.close()
 
    # 删除数据
    def delete_data(self):
        conn = get_db_connection()
        cur = conn.cursor()
        cur.execute("DELETE FROM test WHERE id = 1")
        conn.commit()
        cur.close()
        conn.close()

def make_app():
    return tornado.web.Application([
        (r"/", MainHandler),
    ])
 
if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

4.测试

运行应用访问：

python tornado_demo.py

[root@localhost ~]# curl http://localhost:8888
Database curd begin ... 
[[1, "test_name"]]
Database curd end ...