Flask—SQLAlchemy

App启动类 编写接口部分

# -*- coding: utf-8 -*-
# @Time    : 2022/6/26 20:07
# @Author  : ChowRunFa
# @File    : app.py
# @Software: PyCharm
import datetime
from flask import Flask, jsonify, request, make_response
from flask_sqlalchemy import SQLAlchemy

from models import *
from server import MySQLConfig

app = Flask(__name__)
app.config.from_object(MySQLConfig)
with app.app_context():
	db.init_app(app)
	db.create_all()

@app.route("/house",methods=["GET"])
def getHouse():
		houses = housing.query.all()
		data = []
		for house in houses:
			house = house.__dict__
			del house['_sa_instance_state']
			data.append(house)
		return result(200, {'house':data})

@app.route("/history",methods=["GET"])
def getHistory():
		historires = history.query.all()
		data = []
		for hist in historires:
			hist = hist.__dict__
			del hist['_sa_instance_state']
			data.append(hist)
		return result(200, {'hist':data})


if __name__ == "__main__":
	app.run(debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

数据库表实体类

# -*- coding: utf-8 -*-
# @Time    : 2022/6/26 16:22
# @Author  : ChowRunFa
# @File    : models.py
# @Software: PyCharm
from flask_sqlalchemy import SQLAlchemy
import simplejson as json#pip install simplejson 或者 pip install python-simplejson

db = SQLAlchemy()

def result(code=200,d={}):
	data = dict()#object.__dict__
	data['code'] = code
	data['data'] = d
	return json.dumps(data,ensure_ascii=False)


class housing(db.Model):
	__tablename__ = "housing"
	# __table_args__ = {'mysql_collate':'utf8_general_ci'}

	name = db.Column(db.String(256),primary_key=True)
	addr = db.Column(db.String(256))
	price = db.Column(db.String(256))
	type = db.Column(db.String(256))
	area = db.Column(db.String(256))
	feature = db.Column(db.String(256))


class history(db.Model):
	__tablename__ = "history"
	__table_args__ = {'mysql_collate':'utf8_general_ci'}

	addr = db.Column(db.String(256),primary_key=True)
	time = db.Column(db.String(256),primary_key=True)
	price = db.Column(db.String(256))
	rate = db.Column(db.String(256))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

数据库配置类

# -*- coding: utf-8 -*-
# @Time    : 2022/6/25 11:50
# @Author  : ChowRunFa
# @File    : server.py.py
# @Software: PyCharm
import pymysql
from dbConfig import *
import time
import logging
import traceback
import pymysql.cursors

#mysql settings
class MySQLConfig(object):

	SQLALCHEMY_DATABASE_URI = "mysql+pymysql://{username}:{password}@{ipaddress}:{port}/{database}"\
        .format(username="root",password="123456",ipaddress="127.0.0.1",port="3306",database="jx_db4")
	SQLALCHEMY_TRACK_MODIFICATIONS = True#动态追踪修改设置
	SQLALCHEMY_ECHO = True

# WHITE_NAME_LIST = [""]


class Connection(object):
    """A lightweight wrapper around PyMySQL.
    """
    def __init__(self,max_idle_time=7 * 3600, connect_timeout=10,
                 time_zone="+0:00", charset = "utf8mb4", sql_mode="TRADITIONAL"):
        self.host = MYSQL_HOST
        self.database = MYSQL_DBNAME
        self.max_idle_time = float(max_idle_time)

        args = dict(use_unicode=True, charset=MYSQL_CHARSET,
                    database=MYSQL_DBNAME,
                    init_command=('SET time_zone = "%s"' % time_zone),
                    cursorclass=pymysql.cursors.DictCursor,
                    connect_timeout=connect_timeout, sql_mode=sql_mode)
        args["user"] = MYSQL_USER
        args["passwd"] = MYSQL_PASSWORD
        # We accept a path to a MySQL socket file or a host(:port) string
        args["host"] = MYSQL_HOST
        args["port"] = MYSQL_PORT
        self._db = None
        self._db_args = args
        self._last_use_time = time.time()
        try:
            self.reconnect()
        except Exception:
            logging.error("Cannot connect to MySQL on %s", self.host,
                          exc_info=True)
    def _ensure_connected(self):
        # Mysql by default closes client connections that are idle for
        # 8 hours, but the client library does not report this fact until
        # you try to perform a query and it fails.  Protect against this
        # case by preemptively closing and reopening the connection
        # if it has been idle for too long (7 hours by default).
        if (self._db is None or
            (time.time() - self._last_use_time > self.max_idle_time)):
            self.reconnect()
        self._last_use_time = time.time()

    def _cursor(self):
        self._ensure_connected()
        return self._db.cursor()

    def __del__(self):
        self.close()

    def close(self):
        """Closes this database connection."""
        if getattr(self, "_db", None) is not None:
            self._db.close()
            self._db = None

    def reconnect(self):
        """Closes the existing database connection and re-opens it."""
        self.close()
        self._db = pymysql.connect(**self._db_args)
        self._db.autocommit(True)

    def query(self, query, *parameters, **kwparameters):
        """Returns a row list for the given query and parameters."""
        cursor = self._cursor()
        try:
            cursor.execute(query, kwparameters or parameters)
            result = cursor.fetchall()
            return result
        finally:
            cursor.close()

    def get(self, query, *parameters, **kwparameters):
        """Returns the (singular) row returned by the given query.
        """
        cursor = self._cursor()
        try:
            cursor.execute(query, kwparameters or parameters)
            return cursor.fetchone()
        finally:
            cursor.close()

    def execute(self, query, *parameters, **kwparameters):
        """Executes the given query, returning the lastrowid from the query."""
        cursor = self._cursor()
        try:
            cursor.execute(query, kwparameters or parameters)
            return cursor.lastrowid
        except Exception as e:
            if e.args[0] == 1062:
                pass
            else:
                traceback.print_exc()
                raise e
        finally:
            cursor.close()

    insert = execute

    ## =============== high level method for table ===================

    def table_has(self, table_name, field, value):
        if isinstance(value, str):
            value = value.encode('utf8')
        sql = 'SELECT %s FROM %s WHERE %s="%s"' % (
            field,
            table_name,
            field,
            value)
        d = self.get(sql)
        return d

    def table_insert(self, table_name, item):
        '''item is a dict : key is mysql table field'''
        fields = list(item.keys())
        values = list(item.values())
        fieldstr = ','.join(fields)
        valstr = ','.join(['%s'] * len(item))
        for i in range(len(values)):
            if isinstance(values[i], str):
                values[i] = values[i].encode('utf8')
        sql = 'INSERT INTO %s (%s) VALUES(%s)' % (table_name, fieldstr, valstr)
        try:
            last_id = self.execute(sql, *values)
            return last_id
        except Exception as e:
            if e.args[0] == 1062:
                # just skip duplicated item
                pass
            else:
                traceback.print_exc()
                print('sql:', sql)
                print('item:')
                for i in range(len(fields)):
                    vs = str(values[i])
                    if len(vs) > 300:
                        print(fields[i], ' : ', len(vs), type(values[i]))
                    else:
                        print(fields[i], ' : ', vs, type(values[i]))
                raise e

    def table_update(self, table_name, updates,
                     field_where, value_where):
        '''updates is a dict of {field_update:value_update}'''
        upsets = []
        values = []
        for k, v in updates.items():
            s = '%s=%%s' % k
            upsets.append(s)
            values.append(v)
        upsets = ','.join(upsets)
        sql = 'UPDATE %s SET %s WHERE %s="%s"' % (
            table_name,
            upsets,
            field_where, value_where,
        )
        self.execute(sql, *(values))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175