前言

收到一个简单要求，将GPS平台的实时定位坐标通过JSON接口返回，考虑到功能简单，选择采用Flask轻量化框架+SQLAlchemy调用Mysql数据库，并封装成Docker容器部署上线

技术栈

服务

本页面不提供服务部署教程

Python== 3.8.13

Mysql== 5.7

Docker== 20.10.17

docker-compose== 1.25.1

Python依赖包

Flask== 2.2.2

sqlalchemy== 1.4.42

sqlacodegen== 2.3.0

pymysql== 1.0.2

参数

数据库名: jt808

数据库表名: tgps_car

所需字段:

carno[车牌]

sim[登记卡号]

lat[纬度]

lng[经度]

systime[系统时间]

Flask项目案例

安装依赖包

pip install Flask==2.2.2

pip install sqlalchemy==1.4.42

pip install sqlacodegen==2.3.0

pip install pymysql==1.0.2

sqlacodegen生成SQLAlchemy模型

由于数据库表已存在，可以使用sqlacodegen连接数据库生成模型，解决字段过多问题

sqlacodegen mysql+pymysql://数据库账号:数据库密码@数据库IP地址:3306/jt808 --outfile=models.py  --tables tgps_car

--outfile=`models.py` 将模型输出到`models.py`文件中，默认自动新建

--tables tgps_car 指定生成模型的表，此处填写为`tgps_car`表生成模型

编写Flask框架项目

本项目直接用Pycharm新建Flask框架，会自动生成app.py文件

目录结构

—static/

—templates/

—app.py

—models.py

—Dockerfile

—requirements.txt

—supervisor_flask.conf

app.py

路由为 /search ，请求方式为POST

传参carno或者sim二选一，无法一起传值

效果实现：当carno传参车牌号码或者是sim传参卡号可以返回最新经纬度

from flask import Flask, jsonify, request

from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://数据库账号:数据库密码@数据库IP地址:3306/jt808'

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

app.config['SQLALCHEMY_ECHO'] = True

app.config["JSON_AS_ASCII"] = False

db = SQLAlchemy(app)

# 此处将models.py生成的模型进行复制修改即可，正常只需要将所需字段模型取出，我这边为了后期方便将使用所有模型字段

# 注意：模型字段都需要添加db实例,并且INTEGER类型需要删除指定长度

class JT808(db.Model):

    tablename = "tgps_car"  # 表名 默认使用类名的小写

    # 定义类属性 记录字段

    id = db.Column(db.String(36), primary_key=True)

    carno = db.Column(db.String(100), nullable=False, server_default=db.text("''"), comment='车牌号码')

    sim = db.Column(db.String(64), nullable=False, comment='车载SIM')

    zdid = db.Column(db.String(32))

    ter_id = db.Column(db.INTEGER, nullable=False, comment='终端类型')

    zjh = db.Column(db.String(32), server_default=db.text("''"), comment='车架号')

    czxm = db.Column(db.String(32), server_default=db.text("''"), comment='车主姓名')

    tel = db.Column(db.String(32), server_default=db.text("''"), comment='手机号码')

    pinpai = db.Column(db.String(64), server_default=db.text("''"), comment='品牌')

    online = db.Column(db.CHAR(1), server_default=db.text("'0'"), comment='是否在线')

    bylc = db.Column(db.INTEGER, server_default=db.text("'0'"), comment='保养里程')

    mileage = db.Column(db.DECIMAL(10, 1), comment='总里程')

    stime = db.Column(db.CHAR(14), server_default=db.text("''"), comment='服务开始时间')

    etime = db.Column(db.CHAR(14), server_default=db.text("''"), comment='服务截止时间')

    lat = db.Column(db.DECIMAL(10, 6), comment='纬度')

    lng = db.Column(db.DECIMAL(10, 6), comment='经度')

    speed = db.Column(db.INTEGER, comment='时速')

    direction = db.Column(db.INTEGER, comment='方向')

    location = db.Column(db.CHAR(1), server_default=db.text("'1'"), comment='是否定位')

    acc = db.Column(db.CHAR(1), comment='最后的ACC状态')

    csq = db.Column(db.INTEGER, server_default=db.text("'0'"), comment='信号强度')

    voltage = db.Column(db.Float, server_default=db.text("'0'"), comment='电压')

    gpstime = db.Column(db.CHAR(14), comment='最后定位时间')

    systime = db.Column(db.CHAR(14), comment='系统时间')

    status = db.Column(db.String(2000), server_default=db.text("''"), comment='车辆状态')

    remark = db.Column(db.String(256), comment='备注')

    create_time = db.Column(db.CHAR(14))

    modify_time = db.Column(db.CHAR(14))

    color = db.Column(db.String(64), server_default=db.text("'0'"), comment='车辆颜色')

    sfz = db.Column(db.String(64), comment='身份证')

    oil = db.Column(db.DECIMAL(6, 1), comment='油量')

    height = db.Column(db.INTEGER, comment='高程')

    alarm = db.Column(db.INTEGER)

    jtsjhm = db.Column(db.String(32))

    is_video = db.Column(db.CHAR(1))

# 路由为 /search ，请求方式为POST

# 传参carno或者sim二选一，无法一起传值

#效果实现：当carno传参车牌号码或者是sim传参卡号可以返回最新经纬度

@app.route('/search', methods=['POST'])

def Get_Carno():

    carno = request.form.get("carno")

    sim = request.form.get("sim")

    if carno is not None and sim is not None:

        return jsonify({"code": 0, "message": "carno和sim只能二选一传参"})

    data = None

    ret = {}

    if carno is not None:

        data = JT808.query.filter_by(carno=carno).first()

    if sim is not None:

        data = JT808.query.filter_by(sim=sim).first()

    if data is None:

        ret['status'] = -1

        ret['message'] = '找不到数据'

        ret['data'] = ''

    else:

        ret['status'] = 1

        ret['message'] = 'success'

        ret['data'] = {'carno': data.carno, 'sim': data.sim, 'lat': data.lat, 'lng': data.lng, 'systime': data.systime}

    return jsonify(ret=ret)

if name == '__main__':

    app.run()

Docker封装

supervisor_flask.conf【进程管理】

[supervisord]

nodaemon=true

[program:flaskweb]

command=gunicorn -w 5 -b 0.0.0.0:5000 app:app

directory=/src

autostart=true

autorestart=true

redirect_stderr=true

stdout_logfile=/dev/stdout

gunicorn -w 5 -b 0.0.0.0:5000 app:app

-w 5 表示工作进程数

app:app表示使用app.py启动文件中的app实例，请确认app = Flask(__name__)

Dockerfile

FROM python:3.7-slim

WORKDIR /src

COPY . /src

# 时区

ENV TZ=Asia/Shanghai

RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone

# 替换源

RUN sed -i s/deb.debian.org/mirrors.aliyun.com/g /etc/apt/sources.list \

    && sed -i s/security.debian.org/mirrors.aliyun.com/g /etc/apt/sources.list \

    && apt-get update \

    && apt-get install -y default-libmysqlclient-dev gcc \

    && apt-get clean

RUN mkdir -p ~/.pip && mkdir -p /etc/supervisor

# pip源

RUN echo "[global]" >> ~/.pip/pip.conf \

    && echo "trusted-host=mirrors.aliyun.com" >> ~/.pip/pip.conf \

    && echo "index-url=http://mirrors.aliyun.com/pypi/simple/" >> ~/.pip/pip.conf

# 升级pip

RUN python -m pip install --upgrade pip

# supervisor_conf 进程管理

RUN pip install supervisor flask gunicorn

RUN /usr/local/bin/echo_supervisord_conf > /etc/supervisor/supervisord.conf \

    && echo "[include]">>/etc/supervisor/supervisord.conf \

    && echo "files = /etc/supervisor/*.conf">>/etc/supervisor/supervisord.conf

# 拷贝项目进程文件

COPY ./supervisor_flask.conf /etc/supervisor

RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["supervisord","-n","-c","/etc/supervisor/supervisord.conf"]

requirement.txt

终端执行生成项目依赖包文件

注意请检查里面是否存在本地路径，如有本地路径请直接清除后缀路径即可

pip freeze > requirement.txt

构建docker image镜像

请确保目录结构已存在

—static/

—templates/

—app.py

—models.py

—Dockerfile

—requirements.txt

—supervisor_flask.conf

# 在项目目录中执行Docker镜像构建命令

docker build -t jt808_post:v1 .

镜像名：`jt808_post`

镜像版本：`v1`

docker-compose.yml管理docker

使用docker-compose方便管理docker容器启动和关闭

version: '2'

services:

  jt808-post:

    image: jt808_post:v1

    container_name: jt808-post

    restart: always

    ports:

      - 5000:5000

image: jt808_post:v1 使用镜像`jt808_post:v1`

container_name: jt808-post 容器名称 jt808-post，建议和services一致，可以使用名字管理容器

ports 为映射端口

- 5000:5000即将宿主机5000端口映射为容器的5000端口

启动docker容器

# 需要在docker-compose.yml文件目录下执行

docker-compose up -d jt808-post   # 启动名为jt808-post容器

docker-compose stop jt808-post   # 关闭名为jt808-post容器

docker-compose rm -f jt808-post  # 删除名为jt808-post容器