Python学习day17 任务发布
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Python学习day17 任务发布相关的知识，希望对你有一定的参考价值。
Created on 2017年7月17日
第1课 本节内容 8minutes
任务编排系统开发
架构思路/实现方式介绍
项目实现
接口认证
扩展：
Python的类是什么
模板语言的本质
第2课 任务编排系统架构 46minutes
发任务让机器执行
gitlab
github
svn
git.oschina.net 码云
对ManytoMany追加列，可以在Models字段中添加以下内容：
nane = models.ManytoMany(through=‘table name‘)
一个任务系统的表结构：
UserType Userinfo Admin Usergroup
HostStatus Host TaskTemplate TaskType
ExecuteType Task TaskHoststatus Tasklog
第3课 任务编排后台管理功能介绍一 25minutes
第4课 任务编排后台管理功能介绍二 15minutes
任务后台系统页面书写实现
第5课 任务编排后台管理之任务列表 10minutes
第6课 任务编排后台管理之创建任务一 46minutes
from django import forms, templatetags
from pip._vendor.requests.utils import is_valid_cidr
from django.shortcuts import render_to_response
from cgitb import html
from django.template.backends.django import Template
from _codecs import register
#---------froms-----------------------------------------------
class TaskForm(forms.Form):
name = forms.CharField(max_length=30,
error_messages={‘required‘:u‘任务名称不能为空‘},
widget = forms.TextInput(attrs={‘class‘:
‘form-control no radius‘,‘placeholder‘:
u‘任务名称‘}))
task_type = forms.IntegerField(error_messages={‘required‘:u‘任务类型不能为空‘},
widget=forms.widgets.Select(
choices=models.TaskType.objects.all().order_by(‘id‘).values_list(‘id‘,‘caption‘),
attrs={‘class‘: ‘form-control no radius‘}))
hosts = forms.CharField(error_messages={‘required‘:u‘任务类型不能为空‘},
widget=forms.widgets.SelectMultiple(
choices=models.Hosts.objects.all().order_by(‘id‘).values_list(‘id‘,‘hostname‘),
attrs={‘class‘: ‘form-control no radius‘,‘multiple‘:"multiple"}))
kick_off_time = forms.CharField(max_length=30,
error_messages={‘required‘:u‘执行时间不能为空‘},
widget=forms.DateTimeInput(
attrs={‘class‘: ‘form-control no radius‘,
‘placeholder‘:u‘执行时间‘,‘id‘:‘kick_off_at‘
}))
def __init__(self):
#每执行一次都从数据库中更新，否则每次都是原先数据，新数据不显示
#静态字段执行后会写入内存
self.fields[‘hosts‘].widget.choices = models.Hosts.objects.all().order_by(‘id‘).values_list(‘id‘,‘hostname‘)
#-------------views------------------------------------------------
def add_task(request):
pass
def create(request):
form_obj = TaskForm()
if request.method == ‘POST‘:
form_obj = TaskForm(request.POST)
if form_obj.is_valid:
print form_obj.clean()
raw_data = form_obj.clean()
add_task(raw_data) #将Form的内容添加到数据库中
else:
#显示错误信息
print form_obj.errors.as_data()
return render_to_response(‘index.html‘,
{‘model‘,form_obj,
‘message‘:form_obj.errors.as_data()})
#-------------- 自定义错误模板----------------------------------------
文件目录在APP下，模块名：templatetags,文件名可定义
#form_tag:
form django import template
register = template.Library()
@register.simple_tag
def error_message(arg):
if arg:
return arg[0][0]
else:
return ‘‘
上面模块可显示第一个参数
html使用 {% load form_tag %}
然后可以使用：{% error_message 变量.字段1 %}
第7课 任务编排后台管理之创建任务二 23 minutes
第8课 任务编排后台管理之创建任务三 7 minutes
from django.db import transaction
def add_task(request):
try:
with transaction.atomic():
hosts = data[‘hosts‘]
del data[‘hosts‘]
del data[‘task_template‘]
data[‘task_type‘] = models.TaskType.objects.get(id=data[‘task_type‘])
data[‘execute_type‘] = models.ExecuteType.objects.get(id=data[‘execute_type‘])
task_obj = models.Task.objects.create(**data)
hosts = hosts.replace("‘",‘"‘)
hosts = hosts.replace(‘u‘,‘ ‘)
‘‘‘
hosts = ‘[u"1",u"2",u"3"]‘
Json不能load带U的字符
同时Loads时注意表里面必须是双引号，外面是单引号，否则出错
>>> a
‘[u"1",u"2",u"3"]‘
>>> a = a.replace("u",‘ ‘)
>>> json.loads(a)
[‘1‘, ‘2‘, ‘3‘]
>>> a = ‘[u"1",u"2",u"3"]‘
>>> json.loads(a)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "C:\Python27\lib\json\__init__.py", line 310, in loads
return _default_decoder.decode(s)
File "C:\Python27\lib\json\decoder.py", line 346, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
File "C:\Python27\lib\json\decoder.py", line 364, in raw_decode
raise ValueError("No JSON object could be decoded")
ValueError: No JSON object could be decoded
>>>
‘‘‘
#这里去U改成双引号后才能Loads
host_list = models.Host.objects.filter(id__in=json.loads(hosts))
for item in host_list:
models.TaskHostStatus.objects.create(status=0,task=task_obj,host=item)
第9课 任务编排Agent实现分析 27 minutes
#--------------------------------------------------------------------------
Agent
import json
import uuid
from lib.plugins import PluginApi
from lib.commons import log
from log.core import securty
import config
import commands
class Program:
def __init__(self):
self.host = config.configuations[‘host‘]
self.port = config.configuations[‘port‘]
self.resource = config.configuations[‘resource‘]
self.timeout = config.configuations[‘timeout‘]
def process(self):
data = self.get_task()
#这里可以判断，命令是脚本还是命令，如果是命令直接执行，不需写入文件
#如果是脚本，写入文件再执行
file_name = self.write_file(data)
retsult = self.execute(file_name)
def get_task(self):
params = urllib.urlencode({‘data‘:json.dumps(
{‘hostname‘:c1.salt.com })})
result = self.url_request(params,‘GET‘)
return result
def wirte_file(self):
file_name = str(uuid.uuid())+.‘py‘
f = file(file_name,‘w‘)
f.write(data)
f.close()
return file_name
def execute(self,file_name):
shell_command = ‘python %s‘%(file_name)
status,output = commands.getstatusoutput(shell_command)
pirnt ‘output:==========>‘
print output
def url_request(self,params,methos):
original = None
headers = {‘Content-type‘:
"application/x-www/from-urlencoded","Accept":
"text/json","SecurtyKey":securty.create_ai_key()}
try:
conn = httplib.HTTPConnection(self.host,self.port,self.timeout)
conn.request(metod,self.resource,params,headers)
response = conn.getresponse()
original = response.read()
except Exception,e:
log.write_error_log(‘[htp],%s‘ %e)
return original
第10课 任务编排之API验证 38 minutes
client:
发送 md5(key+datetime)|datetime
server:
1.接收请求 ，分割字符串(加密码，客户端时间）
2.如果当前时间－客户端时间>5s,请求失效
3.md5(客户端时间+key),生成加密码
4.比对加密码是否一致
#--------------------------------------------------------------
#securty -----生成加密串
import time
import config
import hashlib
def create_api_key():
hash_obj = hashlib.md5()
key = config.configration[‘key‘]
time_span = time.time()
hash_obj.updata("%s|%f"%(key,time_span))
encryption = hash_obj.hexdigest()
result = ‘%s|%f‘%(encryption,time_span)
return result
#----------------views----------------------------------------------------
from django.shortcuts import HttpResponse
def api_auth(func):
def wrapper(request):
securty_key = request.META.get(‘HTTP_SECURTYKEY‘,None)
if not securty_key:
return HttpResponse(‘认证失败‘)
if not auth_api_valid(securty_key):
return HttpResponse(‘认证失败‘)
return func(request)
return wrapper
#对接收的数据进行MD5匹配
def auth_api_valid(data):
try:
encryption ,time_span = data.split(‘|‘)
time_span = float(time_span)
if (time.time()-time_span)>5:
return False
hash_obj = hashlib.md5()
hash_obj.update("%s|%f"%(key,time_span))
if hash_obj.hexdigest() = encryption:
return True
else:
return False
except Exception,e:
pass
return False
‘‘‘
这里可以将Key,5等放到一个配置文件里，再直接调用，可方便更改管理
‘‘‘
@api_auth #使用装饰器来检查是否有KEY
def handle_server_info(request):
ret = {‘status‘:0,‘message‘:‘‘}
return HttpResponse(json.dumps(ret))
第11课 扩展之类是什么（上） 34 minutes
第12课 扩展之类是什么（下） 2 minutes
python中一切都是对象，类本身也是对象，类是由type产生的。
class Foo
pass
以下两种方式都是一样的结果
1. Bar = type(‘Bar‘,(object,),{‘name‘:123,‘Func‘:Hello})
2. class Bar:
name = 123
def Hello(self):
print ‘hello‘
既然这样，那么对于定义的类来说，只要定义了一个类，就调用一次 type类的构造函数，如何验证？
__metaclass__ 可以指定类是由那个type来产生的
class MyClass(type):
def __init__(self,name,bases,dicts):
print name
def __call__(self, *args, **kwargs):
return type.__call__(self, *args, **kwargs)
class C1: #是MyClass的实例
__metaclass__ = MyClass
class C2: #是MyClass的实例
__metaclass__ = MyClass
c1 = C1() #执行Call方法
class Test(object):
def __init__(self):
print ‘init‘
def __call__(self):
print ‘call‘
t1 = Test() #执行Init
t1() #执行Call
第13课 扩展之模板语言的本质 23 minutes
模板语言就是通过以下的方式来实现的
#!usr/bin/env python
#coding:utf-8
下面的Name相当于后台发过来的字典变量
namespace = {‘name‘:‘wupeiqi‘,‘data‘:[18,73,84]}
code = ‘‘‘def hellocute():return "name %s ,age %d" %(name,data[0],) ‘‘‘
#通过以下两个命令执行上面的字符串函数
func = compile(code, ‘<string>‘, "exec")
exec func in namespace
result = namespace[‘hellocute‘]()
print result
第14课 总结 15 minutes