coding相关

文章目录

• • 并行任务进度监控
  • python接口服务

---

并行任务进度监控

import psycopg2 as pg
def doDB(sql,flag):  # flag=1是 *** 作增删改，2是查
    conn = pg.connect(database="i_twin", user="postgres", password="123456", host="192.168.1.96",
                      port="5432")
    cur = conn.cursor()
    cur.execute(sql)
    if flag==1:conn.commit()
    if flag==2:res = cur.fetchall()  # 或fetchone()
    cur.close()
    conn.close()
    if flag==2:return res

import socket
# 获取本机计算机名称
hostname = socket.gethostname()
print(hostname)  # LAPTOP-12FVS7AP
# 获取本机ip
ip = socket.gethostbyname(hostname)
print(ip)  # 192.168.150.30

import random
import time
import threading
import os

# 并行任务进度监控，最后启个接口通过id查询进度状态
def st(x):
    print('do',x)
    c=random.random()*30
    time.sleep(c)
    print('done',x)
    s="select * from z_sch where id='%s' "%(id)
    info=doDB(s,2)[0]
    s="update z_sch  set susnum=%s,sch=%s,updatetime=%s where id='%s' "\
      %(info[4]+1,(info[4]+1)/info[3],round(time.time()),id)

    doDB(s,1)


# 开始创建一条数据
id='3'
num=4  # 任务数
t=round(time.time())
# id pid ip num susnum sch state ctime utime
# 唯一id，进程号int，服务器ip，需计算的点数，完成数，sch进度float，state状态，创建时间戳time.time，更新时间戳
# [('1dri32itn2j3r', 234, '12.12.12.12', 5, 4, 44.0, '执行中', 1649322201, 1649322222)]
s="insert into z_sch values ('%s',%s,'%s',%s,%s,%s,'%s',%s,%s) "%(id,os.getpid(),ip,num,0,0,'执行中',t,t)
doDB(s,1)
# exit()

# 主程序 并行
threads = []
for x in range(num):
  thread = threading.Thread(target=st, args=(x,))
  thread.start()
  threads.append(thread)
for thread in threads:
  thread.join()

# 结束把状态改为完成
s="update z_sch  set state='完成' where id='%s' "%(id)
doDB(s,1)

print('Main end')

python接口服务

可以用自带的http.server，nohup python -m http.server 8001 &，http://192.168.6.103:8001/，不过不方便，支持不够（Unsupported method），更好的有wsgi server和wsgi application
fastapi+uvicorn，wsgiref（麻烦）
可以用django、flask

通信数据包的传输、协议的设计，socket、restful
restful规范：get请求数据、post新增数据、put更新数据、delete删除数据
接口没什么特别的规范，能交互、稳定就是好接口

import os
import flask, json
from flask import request

# flask： web框架，通过flask提供的装饰器@server.route()将普通函数转换为服务
server = flask.Flask(__name__)  # 创建一个服务，把当前这个python文件当做一个服务

# server.config['JSON_AS_ASCII'] = False
# @server.route()可以将普通函数转变为服务 登录接口的路径、请求方式
@server.route('/get', methods=['get'])
def doget():
    try:
        with open(os.path.dirname(__file__)+'a.json') as f:
            data=json.load(f)
            print('get data',data) 
            return data
    except FileNotFoundError:
        return {'data':None}


@server.route('/post1', methods=[ 'post']) 
def dopost1():
    # 获取通过url请求传参的json数据
    a=request.data
    if a==b'':return {'no data': 200}
    a=json.loads(a)
    print('asd',a,type(a))
    if a == dict(): return {'data': None}  # 空数据或不传数据

    id=a['id']
    print(id)
    # a1 = request.values.keys() # 传dict而不是json
    # if a1==set():return {'data':None} # 空数据或不传数据
    # ee=dict()
    # for i in a1:
    #     print(i,request.values.getlist(i))  # a ['2', '3']   12 ['zz']  ,get()不全
    #     ee[i]=request.values.getlist(i)
    # print(list(a1),type(a1))

    with open(os.path.dirname(__file__)+'\'+id+'.json','w') as f:  # 写数据
        json.dump(a,f)  # 传参dict，file
        print(os.path.dirname(__file__))
        print('write sus')
    os.system('python cc.py '+id) # 写入结果文件
    with open(os.path.dirname(__file__)+'\'+id+'end.json') as f:  # 读取结果文件，同步返回
        data=json.load(f)
        print(data) 
        return json.dumps(data)

@server.route('/post2', methods=[ 'post']) 
def dopost2():
    # 获取通过url请求传参的json数据
    a=request.data
    if a==b'':return {'no data': 200}
    a=json.loads(a)
    print('asd',a,type(a))
    if a == dict(): return {'data': None}  # 空数据或不传数据

    id=a['id']
    print(id)

    with open(os.path.dirname(__file__)+'\'+id+'.json','w') as f:  # 写数据
        json.dump(a,f)  # 传参dict，file
        print(os.path.dirname(__file__))
        print('write sus')
        
    os.system('python cc.py '+id)  # 阻塞

    # # 子线程，不阻塞
    # import threading
    # def do():
    #     os.system('python cc.py '+id)
    # a = threading.Thread(target=do)
    # a.start()

    return {'suc sent data':200}


if __name__ == '__main__':
    server.run(debug=True, port=8888, host='0.0.0.0')
    # 指定端口、host,0.0.0.0代表不管几个网卡，任何ip都可以访问
    

import requests
import json
url1='http://192.168.150.30:8888/get'
url2='http://192.168.150.30:8888/post2'

ddd={'a':[2,3],12:'zz',22:[3,2,1,2],'id':'asd2'}
# ddd={'a':'zz'}
# ddd={}

a=requests.post(url2,json.dumps(ddd)) # post给数据
# a=requests.post(url2) 
print(a)  # 
print(a.text)  #  {"suc sent data": 200}
# exit()
b=requests.get(url1) # get拿数据
bv=b.json()# 获取数据
print(b)  # 
print(bv,type(bv)) # {'data': None}