BundlerArgumentParser._test_missing_directory() - Code Metrics - drewsonne/awslambdahelper - Measure and Improve Code Quality continuously with Scrutinizer

BundlerArgumentParser._test_missing_directory() A
last analyzed 2017-07-11 14:36 UTC

↳ Parent: BundlerArgumentParser

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	13
rs	9.4285

# -*- coding: utf-8 -*-
"""
Classes encapsulationg the functionality for the lambd
"""
import ConfigParser
import zipfile

import argparse
import glob
import os
import shutil
import sys
import tempfile
import pip
from zipfile import ZipFile


class BundlerArgumentParser(argparse.ArgumentParser):
    """
    Parses command line arguments, and validates the integrity of the file paths and directories provided.
    """

    def __init__(self):
        """
        Add the cli argument schema
        """
        super(BundlerArgumentParser, self).__init__()
        self.add_argument('--directory', help='Path to the directory to bundle for AWS lambda.', required=True)
        self.add_argument('--requirements_name', default='requirements.txt',
                          help='Name of the requirements file in the target directory')

    def _parse_known_args(self, arg_strings, namespace):
        """
        Parse as the parent does, and then optionally raise an ArgumentException is :code:`--send-to-cfn` is missing
        :code:`--owner`.

        :param arg_strings: List of cli argument strings passed to the arg parser.
        :type arg_strings: list
        :param namespace: Namespace object, created by super :py:class:`argparse.ArgumentParser` namespace object.
        :type namespace: argparse.Namespace
        :returns:

                namespace(*argparse.Namespace*)
                    is the super :py:class:`argparse.ArgumentParser` namespace object, with the with the addition of the
                    arguments parse in this class.

                unparsed_args(*list[str]*)
                    are args which were not parsed by this ArgumentParser.
        """
        namespace, unparsed_args = super(BundlerArgumentParser, self)._parse_known_args(arg_strings, namespace)
        namespace.directory = self._full_path(namespace.directory)
        namespace.requirements_path = os.path.join(namespace.directory, namespace.requirements_name)

        directory_missing = self._test_missing_directory(namespace.directory)
        not_a_directory = self._test_not_a_directory(namespace.directory)
        missing_requirements = self._test_missing_requirements(namespace.requirements_path)

        if directory_missing or not_a_directory:
            directory_action = filter(lambda action: '--directory' in action.option_strings, self._actions).pop()
            raise argparse.ArgumentError(directory_action, directory_missing or not_a_directory)

        if missing_requirements:
            requirement_action = filter(lambda action: '--requirements_name' in action.option_strings,
                                        self._actions).pop()
            raise argparse.ArgumentError(requirement_action, missing_requirements)

        return namespace, unparsed_args

    @staticmethod
    def _test_missing_directory(target_directory):
        """
        If the specified directory is missing, return an error message

        :param target_directory: Fully qualified path to test
        :type target_directory: str
        :return: An error message, or False if the requirements file exists.
        :rtype: Union[str,bool]
        """
        if not os.path.exists(target_directory):
            return "Could not find `--directory={dir}`.".format(dir=target_directory)
        return False

    @staticmethod
    def _test_not_a_directory(target_directory):
        """
        If the specified path is not a directory, return an error message

        :param target_directory: Fully qualified path to test
        :type target_directory: str
        :return: An error message, or False if the requirements file exists.
        :rtype: Union[str,bool]
        """
        if not os.path.isdir(target_directory):
            return "`--directory={dir}` is not a directory.".format(dir=target_directory)
        return False

    @staticmethod
    def _test_missing_requirements(requirements_path):
        """
        If the requirements path does not exist, return an error method

        :param requirements_path: Fully qualified path to test.
        :type requirements_path: str
        :return: An error message, or False if the requirements file exists.
        :rtype: Union[str,bool]
        """
        if not os.path.exists(requirements_path):
            return "Could not find requirements file at `{path}`.".format(path=requirements_path)
        return False

    @staticmethod
    def _full_path(dir_):
        """
        Expand any '~', '../', or './' in the dir\_ path.

        :param dir_: A relative, home relative, or absolute path.
        :return: Fully Qualified path
        :rtype: str
        """
        if dir_[0] == '~' and not os.path.exists(dir_):
            dir_ = os.path.expanduser(dir_)
        return os.path.abspath(dir_)


class LambdahelperBundler(object):
    """
    Handler for the cli tool to archive code up for Lambda
    """

    def __init__(self):
        self.target_directory = None
        self.working_directory = None
        self.requirements_path = None

    def run(self, args=None):
        """
        Entrypoint for our bundler cli tool

        :param args: defaults to :py:data:`sys.argv[1:]`
        :return:
        """

        (self.target_directory,
         self.working_directory,
         self.requirements_path) = self.parse_args(args)

        self.copy_lambda_package_files()

        SetupCfgFile(
            os.path.join(self.target_directory, 'setup.cfg'),
            os.path.join(self.working_directory, 'setup.cfg')
        ).load().write()

        pip.main([
            "install",
            "-t", self.working_directory,
            "-r", self.requirements_path
        ])

        temp_archive = DirectoryZipFile(self.working_directory).create_archive().archive_path

        shutil.move(temp_archive, self.target_directory + ".zip")

    def copy_lambda_package_files(self):
        """
        Copy lambda files to working directory.

        :return:
        """
        for file_name in glob.glob(self.target_directory + os.path.sep + "*.py"):
            shutil.copy(file_name, self.working_directory)

        shutil.copy(self.requirements_path, self.working_directory)

    @staticmethod
    def parse_args(args=None):
        """
        Parse the args
        :param args:
        :return:
        """
        parser = BundlerArgumentParser()
        cli_args = parser.parse_args(sys.argv[1:] if args is None else args)
        if not cli_args.directory:
            parser.print_help()
        return (
            cli_args.directory,
            tempfile.mkdtemp(),
            cli_args.requirements_path
        )


class SetupCfgFile(ConfigParser.ConfigParser, object):
    """
    Make sure we have a setup.cfg file with an empty install.prefix for uploading to lambda.
    """

    def __init__(self, setup_cfg, temp_setup_cfg):
        """
        :param setup_cfg: Location of expected path to existing setpu.cfg
        :type setup_cfg: str
        :param temp_setup_cfg: Location of temporary setup.cfg file for use during packaging
        :type temp_setup_cfg: str
        """
        super(SetupCfgFile, self).__init__()
        self.setup_cfg = setup_cfg
        self.temp_setup_cfg = temp_setup_cfg

    def load(self):
        """
        If the existing setup.cfg exists, load it.

        :return:
        :rtype: awslmabdahelper.cli.SetupCfgFile
        """
        if os.path.exists(self.setup_cfg):
            self.read(self.setup_cfg)

        return self

    def write(self):
        """
        Make sure we have an 'install' section, and that the 'prefix' is set to ''.

        :return:
        """
        if 'install' not in self.sections():
            self.add_section('install')
        self.set('install', 'prefix', '')

        with open(self.temp_setup_cfg, 'w') as cfg:
            super(SetupCfgFile, self).write(cfg)


class DirectoryZipFile(ZipFile, object):
    """
    Handles the zipping of an entire directory
    """

    def __init__(self, target):
        zip_destination = target.rstrip(os.path.sep) + '.zip'

        ZipFile.__init__(self, zip_destination, 'w', zipfile.ZIP_DEFLATED)
        self.source_path = target
        self.archive_path = zip_destination

    def create_archive(self):
        """
        Given a target_directory to compress, and a working_directory to place the files in, compress them
        in a zip archive.

        :return:
        """
        self.zipdir(self.source_path, self.source_path.rstrip('/') + '/')

        return self

    def zipdir(self, path, zip_path_prefix):
        """
        Recursively walk our directory path, and add files to the zip archive.

        :param path: Path to walk which contains our files to be added to the zip archive.
        :param zip_path_prefix:
        :return:
        """
        # ziph is zipfile handle
        for root, dirs, files in os.walk(path):
            for file_name in files:
                source_file = os.path.join(root, file_name)
                archive_file = source_file.replace(zip_path_prefix, '')
                print "    Adding file: '" + archive_file + "'"
                self.write(source_file, archive_file)


1			# -- coding: utf-8 --
2			"""
3			Classes encapsulationg the functionality for the lambd
4			"""
5			import ConfigParser
6			import zipfile
7
8			import argparse
9			import glob
10			import os
11			import shutil
12			import sys
13			import tempfile
14			import pip
15			from zipfile import ZipFile
16
17
18			class BundlerArgumentParser(argparse.ArgumentParser):
19			"""
20			Parses command line arguments, and validates the integrity of the file paths and directories provided.
21			"""
22
23			def __init__(self):
24			"""
25			Add the cli argument schema
26			"""
27			super(BundlerArgumentParser, self).__init__()
28			self.add_argument('--directory', help='Path to the directory to bundle for AWS lambda.', required=True)
29			self.add_argument('--requirements_name', default='requirements.txt',
30			help='Name of the requirements file in the target directory')
31
32			def _parse_known_args(self, arg_strings, namespace):
33			"""
34			Parse as the parent does, and then optionally raise an ArgumentException is :code:`--send-to-cfn` is missing
35			:code:`--owner`.
36
37			:param arg_strings: List of cli argument strings passed to the arg parser.
38			:type arg_strings: list
39			:param namespace: Namespace object, created by super :py:class:`argparse.ArgumentParser` namespace object.
40			:type namespace: argparse.Namespace
41			:returns:
42
43			namespace(argparse.Namespace)
44			is the super :py:class:`argparse.ArgumentParser` namespace object, with the with the addition of the
45			arguments parse in this class.
46
47			unparsed_args(list[str])
48			are args which were not parsed by this ArgumentParser.
49			"""
50			namespace, unparsed_args = super(BundlerArgumentParser, self)._parse_known_args(arg_strings, namespace)
51			namespace.directory = self._full_path(namespace.directory)
52			namespace.requirements_path = os.path.join(namespace.directory, namespace.requirements_name)
53
54			directory_missing = self._test_missing_directory(namespace.directory)
55			not_a_directory = self._test_not_a_directory(namespace.directory)
56			missing_requirements = self._test_missing_requirements(namespace.requirements_path)
57
58			if directory_missing or not_a_directory:
59			directory_action = filter(lambda action: '--directory' in action.option_strings, self._actions).pop()
60			raise argparse.ArgumentError(directory_action, directory_missing or not_a_directory)
61
62			if missing_requirements:
63			requirement_action = filter(lambda action: '--requirements_name' in action.option_strings,
64			self._actions).pop()
65			raise argparse.ArgumentError(requirement_action, missing_requirements)
66
67			return namespace, unparsed_args
68
69			@staticmethod
70			def _test_missing_directory(target_directory):
71			"""
72			If the specified directory is missing, return an error message
73
74			:param target_directory: Fully qualified path to test
75			:type target_directory: str
76			:return: An error message, or False if the requirements file exists.
77			:rtype: Union[str,bool]
78			"""
79			if not os.path.exists(target_directory):
80			return "Could not find `--directory={dir}`.".format(dir=target_directory)
81			return False
82
83			@staticmethod
84			def _test_not_a_directory(target_directory):
85			"""
86			If the specified path is not a directory, return an error message
87
88			:param target_directory: Fully qualified path to test
89			:type target_directory: str
90			:return: An error message, or False if the requirements file exists.
91			:rtype: Union[str,bool]
92			"""
93			if not os.path.isdir(target_directory):
94			return "`--directory={dir}` is not a directory.".format(dir=target_directory)
95			return False
96
97			@staticmethod
98			def _test_missing_requirements(requirements_path):
99			"""
100			If the requirements path does not exist, return an error method
101
102			:param requirements_path: Fully qualified path to test.
103			:type requirements_path: str
104			:return: An error message, or False if the requirements file exists.
105			:rtype: Union[str,bool]
106			"""
107			if not os.path.exists(requirements_path):
108			return "Could not find requirements file at `{path}`.".format(path=requirements_path)
109			return False
110
111			@staticmethod
112			def _full_path(dir_):
113			"""
114			Expand any '~', '../', or './' in the dir\_ path.
115
116			:param dir_: A relative, home relative, or absolute path.
117			:return: Fully Qualified path
118			:rtype: str
119			"""
120			if dir_[0] == '~' and not os.path.exists(dir_):
121			dir_ = os.path.expanduser(dir_)
122			return os.path.abspath(dir_)
123
124
125			class LambdahelperBundler(object):
126			"""
127			Handler for the cli tool to archive code up for Lambda
128			"""
129
130			def __init__(self):
131			self.target_directory = None
132			self.working_directory = None
133			self.requirements_path = None
134
135			def run(self, args=None):
136			"""
137			Entrypoint for our bundler cli tool
138
139			:param args: defaults to :py:data:`sys.argv[1:]`
140			:return:
141			"""
142
143			(self.target_directory,
144			self.working_directory,
145			self.requirements_path) = self.parse_args(args)
146
147			self.copy_lambda_package_files()
148
149			SetupCfgFile(
150			os.path.join(self.target_directory, 'setup.cfg'),
151			os.path.join(self.working_directory, 'setup.cfg')
152			).load().write()
153
154			pip.main([
155			"install",
156			"-t", self.working_directory,
157			"-r", self.requirements_path
158			])
159
160			temp_archive = DirectoryZipFile(self.working_directory).create_archive().archive_path
161
162			shutil.move(temp_archive, self.target_directory + ".zip")
163
164			def copy_lambda_package_files(self):
165			"""
166			Copy lambda files to working directory.
167
168			:return:
169			"""
170			for file_name in glob.glob(self.target_directory + os.path.sep + "*.py"):
171			shutil.copy(file_name, self.working_directory)
172
173			shutil.copy(self.requirements_path, self.working_directory)
174
175			@staticmethod
176			def parse_args(args=None):
177			"""
178			Parse the args
179			:param args:
180			:return:
181			"""
182			parser = BundlerArgumentParser()
183			cli_args = parser.parse_args(sys.argv[1:] if args is None else args)
184			if not cli_args.directory:
185			parser.print_help()
186			return (
187			cli_args.directory,
188			tempfile.mkdtemp(),
189			cli_args.requirements_path
190			)
191
192
193			class SetupCfgFile(ConfigParser.ConfigParser, object):
194			"""
195			Make sure we have a setup.cfg file with an empty install.prefix for uploading to lambda.
196			"""
197
198			def __init__(self, setup_cfg, temp_setup_cfg):
199			"""
200			:param setup_cfg: Location of expected path to existing setpu.cfg
201			:type setup_cfg: str
202			:param temp_setup_cfg: Location of temporary setup.cfg file for use during packaging
203			:type temp_setup_cfg: str
204			"""
205			super(SetupCfgFile, self).__init__()
206			self.setup_cfg = setup_cfg
207			self.temp_setup_cfg = temp_setup_cfg
208
209			def load(self):
210			"""
211			If the existing setup.cfg exists, load it.
212
213			:return:
214			:rtype: awslmabdahelper.cli.SetupCfgFile
215			"""
216			if os.path.exists(self.setup_cfg):
217			self.read(self.setup_cfg)
218
219			return self
220
221			def write(self):
222			"""
223			Make sure we have an 'install' section, and that the 'prefix' is set to ''.
224
225			:return:
226			"""
227			if 'install' not in self.sections():
228			self.add_section('install')
229			self.set('install', 'prefix', '')
230
231			with open(self.temp_setup_cfg, 'w') as cfg:
232			super(SetupCfgFile, self).write(cfg)
233
234
235			class DirectoryZipFile(ZipFile, object):
236			"""
237			Handles the zipping of an entire directory
238			"""
239
240			def __init__(self, target):
241			zip_destination = target.rstrip(os.path.sep) + '.zip'
242
243			ZipFile.__init__(self, zip_destination, 'w', zipfile.ZIP_DEFLATED)
244			self.source_path = target
245			self.archive_path = zip_destination
246
247			def create_archive(self):
248			"""
249			Given a target_directory to compress, and a working_directory to place the files in, compress them
250			in a zip archive.
251
252			:return:
253			"""
254			self.zipdir(self.source_path, self.source_path.rstrip('/') + '/')
255
256			return self
257
258			def zipdir(self, path, zip_path_prefix):
259			"""
260			Recursively walk our directory path, and add files to the zip archive.
261
262			:param path: Path to walk which contains our files to be added to the zip archive.
263			:param zip_path_prefix:
264			:return:
265			"""
266			# ziph is zipfile handle
267			for root, dirs, files in os.walk(path):
268			for file_name in files:
269			source_file = os.path.join(root, file_name)
270			archive_file = source_file.replace(zip_path_prefix, '')
271			print " Adding file: '" + archive_file + "'"
272			self.write(source_file, archive_file)
273

drewsonne / awslambdahelper

GitHub Access Token became invalid

BundlerArgumentParser._test_missing_directory() A last analyzed 2017-07-11 14:36 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

BundlerArgumentParser._test_missing_directory() A
last analyzed 2017-07-11 14:36 UTC