python云资源管理_python管理云服务器
如今是云時代，公司買服務器也從傳統的IDC托管到現在的各大云廠商采購 。這里，我們將以阿里云、騰訊云為例實現云服務器實例的獲取。
1、首先部署django環境，然后安裝django drf, 把drf注冊到APPS中
INSTALLED_APPS =[
...'rest_framework',
]
2、在項目下新建一個Python Package命名為apps，settting.py配置路徑， IDE把apps設置成Mark Directory as source root
importosimportsys#Build paths inside the project like this: os.path.join(BASE_DIR, ...)
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(BASE_DIR,'apps'))
3、在apps可以創建app了, 創建python package命名為resources,然后把resources注冊到APPS中
4、在resources下新建個models.py設計數據表，這一步，可以先獲取阿里云和騰訊云的實例返回結果，然后再進行設計。然后把表同步到數據庫
from django.db importmodelsclassCloud(models.Model):
name= models.CharField(max_length=50, verbose_name='云廠商名稱', help_text='云廠商名稱')
code= models.CharField(max_length=20, verbose_name='云廠商名稱', help_text='云廠商名稱')def __str__(self):returnself.nameclassServer(models.Model):
cloud= models.ForeignKey(Cloud, verbose_name='云廠商', help_text='云廠商')
instanceId= models.CharField(max_length=50, db_index=True, verbose_name='實例ID', help_text='實例ID')
instanceName= models.CharField(max_length=50, db_index=True, verbose_name='實例名', help_text='實例名')
instanceType= models.CharField(max_length=50, verbose_name='實例類型', help_text='實例類型')
osName= models.CharField(max_length=32, verbose_name='操作系統', help_text='操作系統')
cpu= models.CharField(max_length=32, verbose_name='cpu核數', help_text='cpu核數')
memory= models.CharField(max_length=32, verbose_name='內存G', help_text='內存G')
createdTime= models.DateTimeField(verbose_name='創建時間',help_text='創建時間', null=True, blank=True)
expiredTime= models.DateTimeField(verbose_name='到期時間', help_text='到期時間', null=True, blank=True)classIp(models.Model):
ip= models.GenericIPAddressField(verbose_name='IP地址', help_text='IP地址', null=True, blank=True)
inner= models.ForeignKey(Server, null=True, related_name='privateIpAddresses', verbose_name='內網IP', help_text='內網IP')
public= models.ForeignKey(Server, null=True, related_name='publicIpAddresses', verbose_name='外網IP', help_text='外網IP')
models.py
5、在 resources新建兩個python package分別命名為aliyun(阿里云)、qcloud(騰訊云), 加下來就需要用到他們的Python SDK了
6、在settings.py中配置他們的secretId、secretKey、regions(區域)
#ALIYUN
ALIYUN_SECRETID = 'xxxxxxxxxxx'ALIYUN_SECRETKEY= 'xxxxxxxxxxxxxxxxx'ALIYUN_REGIONS= ["xxxx"]#QCLOUD
QCLOUD_SECRETID = 'xxxxxxxxxxx'QCLOUD_SECRETKEY= 'xxxxxxxxxxx'QCLOUD_REGIONS= ["xxxx"]
settings.py
阿里云
編輯aliyun.__init__.py
from django.conf importsettingsfrom aliyunsdkcore importclientdefgetClient():for region insettings.ALIYUN_REGIONS:try:returnclient.AcsClient(settings.ALIYUN_SECRETID, settings.ALIYUN_SECRETKEY, region)exceptException as err:print('獲取阿里云client失敗：', err)
aliyun.__init__.py
新建aliyun.ecs.py
importjsonfrom resources.aliyun importgetClientfrom aliyunsdkecs.request.v20140526.DescribeInstancesRequest importDescribeInstancesRequestfrom resources.serializers importServerSerializerdefgetInnerIps(VpcAttributes):return VpcAttributes['PrivateIpAddress']['IpAddress']defgetPlublicIps(EipAddress):
ip_list=[]
ip_list.append(EipAddress['IpAddress'])returnip_listdefsaveInstance(instance):#print(instance)
data ={}
data['cloud'] = 'aliyun'data['instanceId'] = instance['InstanceId']
data['instanceName'] = instance['InstanceName']
data['instanceType'] = instance['InstanceType']
data['osName'] = instance['OSName']
data['cpu'] = instance['Cpu']
data['memory'] = int(instance['Memory'] / 1024)
data['createdTime'] = instance['CreationTime']
data['expiredTime'] = instance['ExpiredTime']
data['innerIps'] = getInnerIps(instance['VpcAttributes'])
data['publicIps'] = getPlublicIps(instance['EipAddress'])
serializer= ServerSerializer(data=data)ifserializer.is_valid():
serializer.save()else:print('阿里云序列化錯誤:', serializer.errors)defgetEcsList():
client=getClient()
request=DescribeInstancesRequest()
request.set_accept_format('json')
request.set_InstanceNetworkType= 'vpc'request.set_PageSize(100)try:
resp=client.do_action(request)
data=json.loads(resp)
instances= data['Instances']['Instance']#saveInstance(instances[0])
print(len(instances))for instance ininstances:
saveInstance(instance)exceptException as err:print('獲取阿里云實例失敗:', err)
aliyun.ecs.py
騰訊云
編輯qcloud.__init__.py
from django.conf importsettingsfrom tencentcloud.common importcredentialfrom tencentcloud.common.exception.tencent_cloud_sdk_exception importTencentCloudSDKExceptiondefgetCredential():try:returncredential.Credential(settings.QCLOUD_SECRETID, settings.QCLOUD_SECRETKEY)exceptTencentCloudSDKException as err:print(err)
qcloud.__init__.py
創建qcloud.cvm.py
importjsonfrom django.conf importsettingsfrom tencentcloud.cvm.v20170312 importcvm_client, modelsfrom tencentcloud.common.exception.tencent_cloud_sdk_exception importTencentCloudSDKExceptionfrom resources.qcloud importgetCredentialfrom resources.serializers importServerSerializerdefgetCvmClient(region):
cred=getCredential()try:returncvm_client.CvmClient(cred, region)exceptTencentCloudSDKException as err:print('獲取騰訊云client:錯誤:', err)defsaveInstance(instance):#print(instance)
data ={}
data['cloud'] = 'qcloud'data['instanceId'] = instance['InstanceId']
data['instanceName'] = instance['InstanceName']
data['instanceType'] = instance['InstanceType']
data['osName'] = instance['OsName']
data['cpu'] = instance['CPU']
data['memory'] = instance['Memory']
data['innerIps'] = instance['PrivateIpAddresses']
data['publicIps'] = instance['PublicIpAddresses']
data['createdTime'] = instance['CreatedTime']
data['expiredTime'] = instance['ExpiredTime']
serializer= ServerSerializer(data=data)ifserializer.is_valid():
serializer.save()else:print('騰訊云序列化錯誤:', serializer.errors)defgetCvmList():for region insettings.QCLOUD_REGIONS:try:
client=getCvmClient(region)
req=models.DescribeInstancesRequest()
resp=client.DescribeInstances(req)
data=json.loads(resp.to_json_string())
instances= data['InstanceSet']for instance ininstances:
saveInstance(instance)exceptException as err:print('獲取騰訊云主機失敗:', err)
qcloud.cvm.py
序列化和反序列化
創建resources.serializers.py
from rest_framework importserializersfrom . models importCloud, Server, IpclassServerSerializer(serializers.Serializer):
id=serializers.ReadOnlyField()
cloud= serializers.PrimaryKeyRelatedField(queryset=Cloud.objects.all(),help_text='云廠商', many=False)
instanceId= serializers.CharField(required=True, help_text='實例ID')
instanceName= serializers.CharField(required=True, help_text='實例名')
instanceType= serializers.CharField(required=True, help_text='實例類型')
osName= serializers.CharField(required=True, help_text='操作系統')
cpu= serializers.CharField(required=True, help_text='CPU核數')
memory= serializers.CharField(required=True, help_text='內存G數')
innerIps= serializers.ListField(help_text='內網IP', write_only=True)
publicIps= serializers.ListField(help_text='外網IP', write_only=True)
createdTime= serializers.DateTimeField(help_text='創建時間')
expiredTime= serializers.DateTimeField(help_text='到期時間')defgetCloudPk(self, code):try:
obj= Cloud.objects.get(code__exact=code)returnobj.idexceptCloud.DoesNotExist:print('云廠商不存在')raise serializers.ValidationError('云廠商不存在')defto_internal_value(self, data):
data['cloud'] = self.getCloudPk(data['cloud'])returnsuper(ServerSerializer, self).to_internal_value(data)defgetInstance(self, instanceId):try:return Server.objects.get(instanceId__exact=instanceId)exceptServer.DoesNotExist:returnNoneexceptException as err:raiseserializers.ValidationError(err.args)defcreate(self, validated_data):
instance= self.getInstance(validated_data['instanceId'])if instance is notNone:returnself.update(instance, validated_data)
innerIps= validated_data.pop('innerIps')
publicIps= validated_data.pop('publicIps')
instance= Server.objects.create(**validated_data)
self.check_inner_ip(innerIps, instance)
self.check_public_ip(publicIps, instance)returninstancedefcheck_inner_ip(self, innerIps, instance):
ip_queryset=instance.privateIpAddresses.all()
current_ip_obj=[]for ip ininnerIps:try:
ip_obj= Ip.objects.get(ip__exact=ip)exceptIp.DoesNotExist:
ip_obj= Ip.objects.create(ip=ip, inner=instance)
current_ip_obj.append(ip_obj)
not_exist_ip= set(ip_queryset) -set(current_ip_obj)for ip_obj innot_exist_ip:
ip_obj.delete()defcheck_public_ip(self, publicIps, instance):
ip_queryset=instance.publicIpAddresses.all()
current_ip_obj=[]for ip inpublicIps:try:
ip_obj= Ip.objects.get(ip__exact=ip)exceptIp.DoesNotExist:
ip_obj= Ip.objects.create(ip=ip, public=instance)
current_ip_obj.append(ip_obj)
not_exist_ip= set(ip_queryset) -set(current_ip_obj)for ip_obj innot_exist_ip:
ip_obj.delete()defupdate(self, instance, validated_data):
instance.instanceName= validated_data.get('instanceName')
instance.osName= validated_data.get('osName')
instance.cpu= validated_data.get('cpu')
instance.memory= validated_data.get('memory')
instance.createdTime= validated_data.get('createdTime')
instance.expiredTime= validated_data.get('expiredTime')
instance.save()
self.check_inner_ip(validated_data['innerIps'], instance)
self.check_public_ip(validated_data['publicIps'], instance)returninstancedefto_representation(self, instance):
ret=super(ServerSerializer, self).to_representation(instance)
ret['cloud'] ={'id': instance.cloud.id,'name': instance.cloud.name
}
ret['innerIps'] = [ip.ip for ip ininstance.privateIpAddresses.all()]
ret['publicIps'] = [ip.ip for ip ininstance.publicIpAddresses.all()]return ret
resources.serializers.py
新建resources.views.py
from django.http importHttpResponsefrom rest_framework importviewsetsfrom django.views importViewfrom resources.qcloud importcvmfrom resources.aliyun importecsfrom . serializers importServerSerializerfrom . models importServerclassQtest(View):def get(self, request, *args, **kwargs):
cvm.getCvmList()return HttpResponse('Qtest')classAtest(View):def get(self, request, *args, **kwargs):
ecs.getEcsList()return HttpResponse('Atest')classServerViewSet(viewsets.ModelViewSet):
queryset=Server.objects.all()
serializer_class= ServerSerializer
resources.views.py
新建resources.router.py
from rest_framework.routers importDefaultRouterfrom . views importServerViewSet
router=DefaultRouter()
router.register('servers', ServerViewSet, base_name='servers')
resources.router
新建resourcrs.urls.py
from django.conf.urls importurlfrom . views importQtest, Atest
urlpatterns=[
url(r'^qtest/', Qtest.as_view(), name='qtest'),
url(r'^atest/', Atest.as_view(), name='atest'),
]
resources.urls.py
編輯根urls.py
from django.conf.urls importurl, includefrom rest_framework.routers importDefaultRouterfrom resources.router importrouter as resources_router
router=DefaultRouter()
router.registry.extend(resources_router.registry)
urlpatterns=[
url(r'^test/', include('resources.urls')),
url(r'', include(router.urls)),
]
urls.py
采集數據
訪問http://127.0.0.1:8000/test/atest 采集阿里云實例數據
訪問http://127.0.0.1:8000/test/qtest采集騰訊云實例數據
任務調度
上面我們是通過http請求來進行數據采集，這只是一個用來測試的方法，在實際應用中，我們可以使用apscheduler任務調度模塊進行定時任務。
安裝django-apscheduler模塊
pip install django-apscheduler
注冊到APPS中
INSTALLED_APPS =[
...'django_apscheduler',
]
django-apscheduler模塊會有兩張表，django_apscheduler_djangojob (定義的任務)，django_apscheduler_djangojobexecution(記錄任務執行情況包括出現的異常)
所以我們需要同步下數據庫
python manage.py makemigrations django_apscheduler
python manage.py migrate apscheduler
新建resources. apscheduler,py
from datetime importdatetimefrom apscheduler.schedulers.background importBackgroundSchedulerfrom django_apscheduler.jobstores importDjangoJobStore, register_job, register_events#建立一個后臺執行的任務
scheduler =BackgroundScheduler()#任務添加存儲
scheduler.add_jobstore(DjangoJobStore(), 'default')#任務調度，每3秒執行一次
@register_job(scheduler, "interval", seconds=3)defmyjob():print('my name is heboan--{}'.format(datetime.now()))#任務執行器
register_events(scheduler)
scheduler.start()
現在只需要在項目根urls.py導入resources. apscheduler即可
from resources import apscheduler
啟動項目后，可以看到它每3秒就執行了一次，有興趣也可以去數據庫查看那兩張表
我們也可以讓任務在前臺運行，這次我們把要執行的任務換成前面的云服務器數據采集。
先把之前根urls.py那個導入模塊去掉
編輯resources. apscheduler,py
from datetime importdatetimefrom apscheduler.schedulers.background importBackgroundSchedulerfrom django_apscheduler.jobstores importDjangoJobStore, register_job, register_eventsfrom resources.qcloud.cvm importgetCvmListfrom resources.aliyun.ecs importgetEcsListfrom apscheduler.schedulers.blocking importBlockingScheduler#scheduler = BackgroundScheduler()#建立前臺任務
scheduler =BlockingScheduler()
scheduler.add_jobstore(DjangoJobStore(),'default')
@register_job(scheduler,"interval", seconds=30)defsync_cloud():
getEcsList()#采集阿里云
getCvmList() #采集騰訊云
register_events(scheduler)
在resources下新建Python Package命名為management
在management下新建Python Package命名為commands
在 commands下新建文件schedule.py
現在我們執行下python manage.py，可以看到多出來一條可以執行的命令
現在，我們來編輯schedule.py
from django.core.management.base importBaseCommandfrom resources.apscheduler importschedulerclassCommand(BaseCommand):
help= "django schedule"
def handle(self, *args, **options):
scheduler.start()
最后我們就可以執行python manage.py schedule了
前臺任務：只需要保持python manage.py schedule在運行就可以了，后臺任務需要項目跑起來?？茨阆矚g怎么用了 ^_^
分頁
當我們訪問http://127.0.0.1:8000/servers/ 會把所有的實例都顯示出來，比較好的做法是進行分頁顯示，最簡單的方式就在settings.py種添加如下配置：
REST_FRAMEWORK ={'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.LimitOffsetPagination','PAGE_SIZE': 10 #每頁顯示的條數
}
這樣整個全局就生效了，按每頁顯示10條進行分頁
當然，我們也可以自定義分頁，如下：
1、在項目(devops)下新建個文件paginations.py
from rest_framework.pagination importPageNumberPaginationclassPagination(PageNumberPagination):
page_size= 10 #每頁顯示的條數
page_size_query_param = 'page_size' #想要顯示的頁數
page_query_param = 'p' #指定頁碼的參數
max_page_size = 100 每頁顯示的最大條數
2、修改settings.py中的分頁設置(這個也會導致全局生效)
REST_FRAMEWORK ={'DEFAULT_PAGINATION_CLASS': 'devops.paginations.Pagination', #這里配置成我們自定義的
'PAGE_SIZE': 10}
上面的配置完成后，就會全局使用我們自定義的分頁配置，如果只只想要Servers使用自定義的分頁配置，那么我們就不需要修改settings.py，只需要如下操作
...from devops.paginations importPaginationclassServerViewSet(viewsets.ModelViewSet):
queryset=Server.objects.all()
serializer_class=ServerSerializer
pagination_class= Pagination #指定自定義的分頁類即可
搜索
服務器實例比較多，如果我們想要搜索出指定的服務器實例，比如根據instanceName來搜索(http://127.0.0.1:8000/servers/?instanceName=pr-es-01)
現在我們在瀏覽器訪問http://127.0.0.1:8000/servers/?instanceName=pr-es-01， 會發現并沒有做任何過濾!
那么我們該怎么辦？
1、我們進行搜索其實就是對模型進行條件篩選
2、在drf中，就是對queryset進行操作了
classServerViewSet(viewsets.ModelViewSet):
queryset=Server.objects.all()
serializer_class=ServerSerializer
pagination_class=Paginationdefget_queryset(self):
queryset=super(ServerViewSet, self).get_queryset()#獲取到instanceName參數
instanceName = self.request.query_params.get('instanceName')#如果有傳遞此參數則進行instanceName模糊查詢
ifinstanceName:
queryset= queryset.filter(instanceName__icontains=instanceName)return queryset
但是一般我們不會使用上面的方法去實現搜索。我們會使用django-filter
當然首先必須要先安裝
pip install django-filter
然后注冊到APPS中
INSTALLED_APPS =[
...'django_filters',
]
然后可以在視圖中使用了
...from django_filters.rest_framework importDjangoFilterBackendclassServerViewSet(viewsets.ModelViewSet):
queryset=Server.objects.all()
serializer_class=ServerSerializer
pagination_class=Pagination#使用過濾器
filter_backends =(DjangoFilterBackend,)
filter_fields= ("instanceName",)
為了支持模糊查詢或其他更高級的查詢，我們就需要自定義過濾器了
新建文件resources.filters.py
from django_filters.rest_framework importFilterSetimportdjango_filtersfrom resources.models importServerclassServerFilter(FilterSet):
instanceName= django_filters.CharFilter(lookup_expr="icontains") #支持模糊查詢
classMeta:
model=Server
fields= ['instanceName']
然后在視圖中指定過濾類就可以了
from resources.filters importServerFilterclassServerViewSet(viewsets.ModelViewSet):
queryset=Server.objects.all()
serializer_class=ServerSerializer
pagination_class=Pagination
filter_backends= (DjangoFilterBackend,) #可以放到全局配置中
filter_class = ServerFilter
上面我提到filter_backends = (DjangoFilterBackend,)可以放到全局配置中，這樣視圖中就不需要配置了。操作如下：
1、先把視圖中這段配置去掉 ： filter_backends = (DjangoFilterBackend,)
2、編輯settings.py
REST_FRAMEWORK ={'DEFAULT_PAGINATION_CLASS': 'devops.paginations.Pagination','PAGE_SIZE': 10,'DEFAULT_FILTER_BACKENDS': ('django_filters.rest_framework.DjangoFilterBackend',) #加上此配置即可
}
ffilter還可以進行一些高級的方法,比如可以查詢多個字段匹配
importdjango_filtersfrom django.contrib.auth importget_user_modelfrom django.db.models importQ
User=get_user_model()classUserFilter(django_filters.rest_framework.FilterSet):"""用戶過濾類"""username= django_filters.CharFilter(method='search_username')defsearch_username(self, queryset, name, value):return queryset.filter(Q(name__icontains=value)|Q(username__icontains=value))classMeta:
model=User
fields= ['username']
與50位技術專家面對面20年技術見證，附贈技術全景圖總結
以上是生活随笔為你收集整理的python云资源管理_python管理云服务器的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： python函数必背知识点_必背函数——
- 下一篇： tomcat 外网访问不了_免费云服务器