用本地标签打印机 自动打印订单
文件结构
sql_util.py
# 导包
import pymysql
class SQLTool:
def __init__(self, user, pwd, host, port, db):
self.client = pymysql.connect(
user=user,
password=pwd,
host=host,
port=port,
database=db
)
self.cursor = self.client.cursor()
def query(self, sql):
# 调用原生sql语句
self.cursor.execute(sql)
aaa = self.fetch_all_to_dict(self.cursor)
return aaa
def fetch_all_to_dict(self, cursor):
print(cursor.description)
desc = [i[0] for i in cursor.description]
result = [dict(zip(desc, col)) for col in cursor.fetchall()]
return result
import datetime
from application import app, db
from common.models.pay.PayOrder import PayOrder
class JobTask():
"""
python manager.py runjob -m mytask/index
"""
def __init__(self):
pass
def run(self, params):
from .sql_util import SQLTool
# 调用SQLTool这个类
tool = SQLTool(
"root",
"123456",
"127.0.0.1",
3306,
"food_db"
)
# print(tool)
# 原生sql语句
sql = "select * from pay_order;"
# 调用类里的方法
res = tool.query(sql)
print(res)
return "ok"
执行
python manager.py runjob -m mytask/index
成功执行 任务且 打印了订单信息
效果图
这个是我的自动5秒 调用标签打印机的代码 且目前能5秒打印一次
print.py
from flask import Flask
from flask_apscheduler import APScheduler
import tempfile
import win32api
import win32print
class Config(object): # 创建配置用类
# 任务列表
JOBS = [
# { # 第一个任务
# 'id': 'job1',
# 'func': '__main__:job_1',
# 'args': (1, 2),
# 'trigger': 'cron', # cron表示定时任务
# 'hour': 19,
# 'minute': 27
# },
{ # 第二个任务每隔5S执行一次
'id': 'job2',
'func': '__main__:method_test', # 方法名
'args': (1, 2), # 入参
'trigger': 'interval', # interval表示循环任务
'seconds': 5,
}
]
def method_test(a, b):
filename = tempfile.mktemp("订单信息.txt")
open(filename, "w").write("
"
"订单编号2019071913272800037
"
"会员姓名坚持下去
"
"订单总价128.60
"
"订单状态待发货
"
"创建时间2019-07-19 13:27:28
"
"收货地址15821063464上海市徐汇区虹桥路693弄11号楼104
"
""
"商品 数量 金额
"
"橙子 1 8.80
"
"鲜活小龙虾 1 75.00
"
"基围虾 1 28.90
"
"猪肉精肉 1 15.90
"
""
"")
win32api.ShellExecute(0, "printto", filename, '"%s"' % win32print.GetDefaultPrinter(), ".", 0)
app = Flask(__name__)
app.config.from_object(Config()) # 为实例化的flask引入配置
##
@app.route("/hello", methods=["POST", "GET"])
def check():
return "success", 200
if __name__ == '__main__':
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()
app.run(debug=False)
我想实现的是 我本地能自动获取阿里云服务器上数据库的数据调用本地标签打印机打印订单信息
from flask import Flask
from flask_apscheduler import APScheduler
import tempfile
import win32api
import win32print
class Config(object): # 创建配置用类
# 任务列表
JOBS = [
# { # 第一个任务
# 'id': 'job1',
# 'func': '__main__:job_1',
# 'args': (1, 2),
# 'trigger': 'cron', # cron表示定时任务
# 'hour': 19,
# 'minute': 27
# },
{ # 第二个任务每隔5S执行一次
'id': 'job2',
'func': '__main__:method_test', # 方法名
'args': (1, 2), # 入参
'trigger': 'interval', # interval表示循环任务
'seconds': 5,
}
]
def method_test(a, b):
# 我想在这里获取 阿里云服务器上数据库的新下的订单信息 ~~~~~~~
# start ~~~~~~~
from .sql_util import SQLTool
# 调用SQLTool这个类
tool = SQLTool(
"root",
"123456",
"127.0.0.1",
3306,
"food_db"
)
# print(tool)
# 原生sql语句
sql = "select * from pay_order;"
# 调用类里的方法
res = tool.query(sql)
print(res)
return "ok"
# end ~~~~~~~
filename = tempfile.mktemp("订单信息.txt")
open(filename, "w").write("
"
"订单编号2019071913272800037
"
"会员姓名坚持下去
"
"订单总价128.60
"
"订单状态待发货
"
"创建时间2019-07-19 13:27:28
"
"收货地址15821063464上海市徐汇区虹桥路693弄11号楼104
"
""
"商品 数量 金额
"
"橙子 1 8.80
"
"鲜活小龙虾 1 75.00
"
"基围虾 1 28.90
"
"猪肉精肉 1 15.90
"
""
"")
win32api.ShellExecute(0, "printto", filename, '"%s"' % win32print.GetDefaultPrinter(), ".", 0)
app = Flask(__name__)
app.config.from_object(Config()) # 为实例化的flask引入配置
##
@app.route("/hello", methods=["POST", "GET"])
def check():
return "success", 200
if __name__ == '__main__':
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()
app.run(debug=False)
报错
C:UsershuangAppDataLocalProgramsPythonPython36python.exe "C:Program FilesJetBrainsPyCharm 2017.2.7helperspydevpydevd.py" --multiproc --qt-support=auto --client 127.0.0.1 --port 55343 --file C:/Users/huang/vincent/code/mooc/order/jobs/tasks/mytask/print.py
pydev debugger: process 10900 is connecting
Connected to pydev debugger (build 172.4574.37)
* Serving Flask app "print" (lazy loading)
* Environment: production
WARNING: This is a development server. Do not use it in a production deployment.
Use a production WSGI server instead.
* Debug mode: off
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
Job "job2 (trigger: interval[0:00:05], next run at: 2019-08-14 14:56:51 CST)" raised an exception
Traceback (most recent call last):
File "C:UsershuangAppDataLocalProgramsPythonPython36libsite-packagesapschedulerexecutorsase.py", line 125, in run_job
retval = job.func(*job.args, **job.kwargs)
File "C:/Users/huang/vincent/code/mooc/order/jobs/tasks/mytask/print.py", line 33, in method_test
from .sql_util import SQLTool
ModuleNotFoundError: No module named '__main__.sql_util'; '__main__' is not a package
Process finished with exit code 1
接下来 我就不知道怎么处理了
625
收起
正在回答 回答被采纳积分+3
1回答
编程浪子
2019-08-14 20:25:59
相似问题
登录后可查看更多问答，登录/注册
