Python: модуль Fabric — пример использования

Автор: | 06/27/2015
 

PythonОбщие сведения о модуле можно найти в посте Python: модуль Fabric.

Ниже — пример использования.

Класс используется для работы с нашей базой данных Cassandra, что бы загружать, получать или удалять файлы.

#!/usr/bin/env python

import os
import sys
import requests
import time
import re

from lib.external.external import Logger

"""To make GET, PUD and DELETE requests to Application Cloudlibrary via REST API"""


class RdsClc(object):

    """To make GET, PUD and DELETE requests to Application Cloudlibrary via REST API"""

    def __init__(self, clc_user, clc_password):

        self.logger = Logger(os.path.dirname(os.path.realpath(__file__)))
        self.logger.logger(__name__)
        self.logger.logger.info('RDS Cloudlibrary module started at %s' % (time.strftime('%d, %b %Y at %H:%M:%S')))

        self.session = requests.Session()
        self.session.auth = (clc_user, clc_password)
        self.session.headers.update({'user-agent': 'RDSmanager'})

    def cloud_get(self, url, outfile=None):

        """Make API request with GET
           For testing purpose only"""

        # checking connection
        try:
            data_check = self.session.get(url)
            data_check.raise_for_status()
            self.logger.logger.info('Connection OK')
        except requests.exceptions.ConnectionError as error:
            self.logger.logger.info('ERROR: %s' % error)
            sys.exit(5)

        if not outfile:
            outfile = re.findall('.*/(.*)', url)[0]

        self.logger.logger.info('Writing %s to outfile %s' % (url, outfile))
        try:
            with open(outfile, 'wb') as o_file:
                data_get = self.session.get(url, stream=True)
                for chunk in data_get.iter_content(65536):
                    o_file.write(chunk)
                data_get.raise_for_status()
                self.logger.logger.info('Done. Result code: %s' % data_get.status_code)
        except requests.exceptions.HTTPError as error:
            self.logger.logger.info('ERROR: %s' % error)
            sys.exit(7)

    def cloud_put(self, url, deploy_file, deploy_name=None):

        """Make API request with PUT
           Executes session.put() for URL and FILE
           Used in lib.unity.deploy.deploy_ci()"""

        # m_url - URL with UUID, e.g. https://www.dev.domain.com/cloudlibrary/data/4ac11ad6-a391-455c-bd6b-d205b655926b
        # m_file - file on local filesystem to deploy, e.g. d:RDSbuilddirPlugins_xmlsProductGroupEditor.unity3d
        # m_name - just module name, e.g. UnityUI
        self.logger.logger.info('Deploying file: %snURL: %s File: %s' % (deploy_name, url, deploy_file))
        try:
            with open(deploy_file, 'rb') as infile:
                data_put = self.session.put(url, data=infile)
                data_put.raise_for_status()
                self.logger.logger.info('Result code: %sn' % data_put.status_code)
        except requests.exceptions.HTTPError as error:
            self.logger.logger.info('ERROR: %s' % error)
            sys.exit(7)
        except requests.exceptions.TooManyRedirects:
            self.logger.logger.info('Authentication failed!')
            sys.exit(6)

    def cloud_delete(self, url):

        """Delete object from Cloudlibrary"""

        try:
            data_delete = self.session.delete(url)
            data_delete.raise_for_status()
            self.logger.logger.info('Result code: %sn' % data_delete.status_code)
        except requests.exceptions.HTTPError as error:
            self.logger.logger.info('ERROR: %s' % error)
            sys.exit(7)
        except requests.exceptions.TooManyRedirects:
            self.logger.logger.info('Authentication failed!')
            sys.exit(6)

Вызывается он из основного скрипта приложения, котрый описан в посте Python: модуль argparse – использование субкоманд.

Например — выполнение GET:

...
def handler_clc(action_list):

    """Check Cloudlibrary namespace options"""

    [...]

    if action_list.clc_get:
        from lib.external.rds_clc import RdsClc
        if not action_list.clc_user and not action_list.clc_password:
            logger.logger.info('Nether CLC_USER not CLC_PASSWORD found, using default values')
            action_list.clc_user = CLC_USER
            action_list.clc_password = CLC_PASSWORD
        req = RdsClc(action_list.clc_user, action_list.clc_password)
        req.cloud_get(action_list.clc_get, action_list.clc_file)
    [...]

И пример работы:

d:RDSrdsmanager>RDSmanager.py clc -g https://www.dev.domain.com/cloudlibrary/data/32fcf895-5e21-415d-9657-55361a1e03f8
RDSmanager started at 25, Jun 2015 at 18:43:14
Nether CLC_USER not CLC_PASSWORD found, using default values
RDS Cloudlibrary module started at 25, Jun 2015 at 18:43:18
Connection OK
Writing https://www.dev.domain.com/cloudlibrary/data/32fcf895-5e21-415d-9657-55361a1e03f8 to outfile 32fcf895-5e21-415d-9657-55361a1e03f8
Done. Result code: 200
d:RDSrdsmanager>RDSmanager.py clc -g https://www.dev.domain.com/cloudlibrary/data/32fcf895-5e21-415d-9657-55361a1e03f8 -f filefromCassandra.dll
RDSmanager started at 25, Jun 2015 at 18:43:36
Nether CLC_USER not CLC_PASSWORD found, using default values
RDS Cloudlibrary module started at 25, Jun 2015 at 18:43:36
Connection OK
Writing https://www.dev.domain.com/cloudlibrary/data/32fcf895-5e21-415d-9657-55361a1e03f8 to outfile filefromCassandra.dll
Done. Result code: 200