Tornado

1.簡介

要在 Tornado 中使用 HighGoDB ，你需要安裝一個 Python 的 HighGoDB 驅動，通常使用 psycopg2 。

本文主要介紹 Tornado 該如何連接使用瀚高數(shù)據(jù)庫。

2.加載psycopg2

參考 Psycopg2 接口文檔。

3.連接瀚高CURD

tornado_demo.py

# -*- coding: utf-8 -*-
import tornado.ioloop
import tornado.web
import psycopg2
import json
 
# 連接數(shù)據(jù)庫配置
DATABASE_CONFIG = {
    "host": "192.168.100.101",
    "dbname": "highgo",
    "user": "sysdba",
    "password": "Qwer@1234",
    "port": "5866",
}
 
# 連接數(shù)據(jù)庫
def get_db_connection():
    return psycopg2.connect(**DATABASE_CONFIG)
 
class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("Database curd begin ... \n")
        # 創(chuàng)建表
        self.create_table()
 
        # 插入數(shù)據(jù)
        self.insert_data()
 
        # 查詢數(shù)據(jù)
        res =  self.select_data()
        self.write(json.dumps(res))
        self.write("\n")
        
        # 更新數(shù)據(jù)
        self.update_data()
 
        # 刪除數(shù)據(jù)
        self.delete_data() 
        
        self.write("Database curd end ... \n")
    
    # 創(chuàng)建表
    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()
 
    # 插入數(shù)據(jù)
    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()
 
    # 查詢數(shù)據(jù)
    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
 
    # 更新數(shù)據(jù)
    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()
 
    # 刪除數(shù)據(jù)
    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 ...