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

以上是关于Python学习day17 任务发布的主要内容,如果未能解决你的问题,请参考以下文章

python学习点滴记录-Day17-django-part2

Python入门学习-DAY37-进程池与线程池协程gevent模块

《30天自制操作系统》17_day_学习笔记

PYTHON学习笔记-DAY-6

python学习day——17

python学习day17