DeleteDirectory.__init__() - Code Metrics - Inspection of "Bump version (+pep8)" - jacquev6/ActionTree - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 31c7ae...e8f49f )

by Vincent

created 2017-06-17 15:00 UTC

DeleteDirectory.init() A

↳ Parent: DeleteDirectory

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# coding: utf8

# Copyright 2013-2017 Vincent Jacques <[email protected]>

"""
Stock actions are predefined common tasks (manipulating the filesystem, calling an external program, etc.)
They all specialize :class:`.Action`.
"""

from __future__ import division, absolute_import, print_function

import errno
import os
import shutil
import subprocess
import time


from . import Action


DEFAULT = object()


class NullAction(Action):
    """
    A stock action that does nothing.
    Useful as a placeholder for several dependencies.
    """
    def __init__(self, label=None, *args, **kwds):
        """
        @todoc
        """
        Action.__init__(self, label, *args, **kwds)

    def do_execute(self, dependency_statuses):
        pass


class CallSubprocess(Action):
    """
    A stock action that calls a subprocess.

    Note: if the process fails,
    :func:`~subprocess.check_call` raises a :exc:`subprocess.CalledProcessError`,
    which `cannot be pickled <http://bugs.python.org/issue1692335>`__ in Python 2.
    So, in that case, this action catches the original exception and
    raises a :exc:`CalledProcessError`.
    """
    def __init__(self, command, kwargs={}, label=DEFAULT, *args, **kwds):
        """
        @todoc
        """
        Action.__init__(self, " ".join(command) if label is DEFAULT else label, *args, **kwds)
        self.__command = command
        self.__kwargs = kwargs

    def do_execute(self, dependency_statuses):
        # subprocess.CalledProcessError can't be pickled in Python2
        # See http://bugs.python.org/issue1692335
        try:
            subprocess.check_call(self.__command, **self.__kwargs)
        except subprocess.CalledProcessError as e:  # Not doctested: implementation detail
            raise CalledProcessError(e.returncode, e.cmd, e.output)


class CalledProcessError(Exception):
    """
    Raised by :class:`CallSubprocess`
    """


class CreateDirectory(Action):
    """
    A stock action that creates a directory.
    No error will be raised if the directory already exists.
    If the directory to create is nested, intermediate directories will be created as well.

    :param str name: the directory to create, passed to :func:`os.makedirs`.
    """
    def __init__(self, name, label=DEFAULT, *args, **kwds):
        """
        @todoc
        """
        Action.__init__(self, "mkdir {}".format(name) if label is DEFAULT else label, *args, **kwds)
        self.__name = name

    def do_execute(self, dependency_statuses):
        try:
            os.makedirs(self.__name)
        except OSError as e:  # Not doctested: implementation detail
            if e.errno != errno.EEXIST or not os.path.isdir(self.__name):
                raise


class DeleteFile(Action):  # Not doctested: could be
    """
    A stock action that deletes a file.
    No error will be raise if the file doesn't exist.

    :param str name: the name of the file to delete, passed to :func:`os.unlink`.
    """
    def __init__(self, name, label=DEFAULT, *args, **kwds):
        """
        @todoc
        """
        Action.__init__(self, "rm {}".format(name) if label is DEFAULT else label, *args, **kwds)
        self.__name = name

    def do_execute(self, dependency_statuses):
        try:
            os.unlink(self.__name)
        except OSError as e:
            if e.errno != errno.ENOENT:
                raise


class DeleteDirectory(Action):  # Not doctested: could be
    """
    A stock action that deletes a directory (recursively).
    No error will be raise if the directory doesn't exist.

    :param str name: the name of the directory to delete, passed to :func:`shutil.rmtree`.
    """
    def __init__(self, name, label=DEFAULT, *args, **kwds):
        """
        @todoc
        """
        Action.__init__(self, "rm -r {}".format(name) if label is DEFAULT else label, *args, **kwds)
        self.__name = name

    def do_execute(self, dependency_statuses):
        try:
            shutil.rmtree(self.__name)
        except OSError as e:
            if e.errno != errno.ENOENT:
                raise


class CopyFile(Action):  # Not doctested: could be
    """
    A stock action that copies a file. Arguments are passed to :func:`shutil.copy`.

    :param str src: the file to copy
    :param str dst: the destination
    """
    def __init__(self, src, dst, label=DEFAULT, *args, **kwds):
        """
        @todoc
        """
        Action.__init__(self, "cp {} {}".format(src, dst) if label is DEFAULT else label, *args, **kwds)
        self.__src = src
        self.__dst = dst

    def do_execute(self, dependency_statuses):
        shutil.copy(self.__src, self.__dst)


class TouchFile(Action):  # Not doctested: could be
    """
    A stock action that touches a file.
    If the file already exists, its modification time will be modified.
    Else, it will be created, empty.

    Note that the containing directory must exist.
    You might want to ensure that by adding a :class:`CreateDirectory` as a dependency.

    :param str name: the name of the file to touch. Passed to :func:`open` and/or :func:`os.utime`.
    """

    def __init__(self, name, label=DEFAULT, *args, **kwds):
        """
        @todoc
        """
        Action.__init__(self, "touch {}".format(name) if label is DEFAULT else label, *args, **kwds)
        self.__name = name

    def do_execute(self, dependency_statuses):
        open(self.__name, "ab").close()  # Create the file if needed
        os.utime(self.__name, None)  # Actually change its time


class Sleep(Action):
    """
    A stock action that sleeps for a certain duration.

    :param float secs: seconds to sleep, passed to :func:`time.sleep`.
    """
    def __init__(self, secs, label=DEFAULT, *args, **kwds):
        """
        @todoc
        """
        Action.__init__(self, "sleep {}".format(secs) if label is DEFAULT else label, *args, **kwds)
        self.__secs = secs

    def do_execute(self, dependency_statuses):
        time.sleep(self.__secs)


1			# coding: utf8
2
3			# Copyright 2013-2017 Vincent Jacques <[email protected]>
4
5			"""
6			Stock actions are predefined common tasks (manipulating the filesystem, calling an external program, etc.)
7			They all specialize :class:`.Action`.
8			"""
9
10			from __future__ import division, absolute_import, print_function
11
12			import errno
13			import os
14			import shutil
15			import subprocess
16			import time
17
18
19			from . import Action
20
21
22			DEFAULT = object()
23
24
25			class NullAction(Action):
26			"""
27			A stock action that does nothing.
28			Useful as a placeholder for several dependencies.
29			"""
30			def __init__(self, label=None, args, *kwds):
31			"""
32			@todoc
33			"""
34			Action.__init__(self, label, args, *kwds)
35
36			def do_execute(self, dependency_statuses):
37			pass
38
39
40			class CallSubprocess(Action):
41			"""
42			A stock action that calls a subprocess.
43
44			Note: if the process fails,
45			:func:`~subprocess.check_call` raises a :exc:`subprocess.CalledProcessError`,
46			which `cannot be pickled <http://bugs.python.org/issue1692335>`__ in Python 2.
47			So, in that case, this action catches the original exception and
48			raises a :exc:`CalledProcessError`.
49			"""
50			def __init__(self, command, kwargs={}, label=DEFAULT, args, *kwds):
51			"""
52			@todoc
53			"""
54			Action.__init__(self, " ".join(command) if label is DEFAULT else label, args, *kwds)
55			self.__command = command
56			self.__kwargs = kwargs
57
58			def do_execute(self, dependency_statuses):
59			# subprocess.CalledProcessError can't be pickled in Python2
60			# See http://bugs.python.org/issue1692335
61			try:
62			subprocess.check_call(self.__command, **self.__kwargs)
63			except subprocess.CalledProcessError as e: # Not doctested: implementation detail
64			raise CalledProcessError(e.returncode, e.cmd, e.output)
65
66
67			class CalledProcessError(Exception):
68			"""
69			Raised by :class:`CallSubprocess`
70			"""
71
72
73			class CreateDirectory(Action):
74			"""
75			A stock action that creates a directory.
76			No error will be raised if the directory already exists.
77			If the directory to create is nested, intermediate directories will be created as well.
78
79			:param str name: the directory to create, passed to :func:`os.makedirs`.
80			"""
81			def __init__(self, name, label=DEFAULT, args, *kwds):
82			"""
83			@todoc
84			"""
85			Action.__init__(self, "mkdir {}".format(name) if label is DEFAULT else label, args, *kwds)
86			self.__name = name
87
88			def do_execute(self, dependency_statuses):
89			try:
90			os.makedirs(self.__name)
91			except OSError as e: # Not doctested: implementation detail
92			if e.errno != errno.EEXIST or not os.path.isdir(self.__name):
93			raise
94
95
96			class DeleteFile(Action): # Not doctested: could be
97			"""
98			A stock action that deletes a file.
99			No error will be raise if the file doesn't exist.
100
101			:param str name: the name of the file to delete, passed to :func:`os.unlink`.
102			"""
103			def __init__(self, name, label=DEFAULT, args, *kwds):
104			"""
105			@todoc
106			"""
107			Action.__init__(self, "rm {}".format(name) if label is DEFAULT else label, args, *kwds)
108			self.__name = name
109
110			def do_execute(self, dependency_statuses):
111			try:
112			os.unlink(self.__name)
113			except OSError as e:
114			if e.errno != errno.ENOENT:
115			raise
116
117
118			class DeleteDirectory(Action): # Not doctested: could be
119			"""
120			A stock action that deletes a directory (recursively).
121			No error will be raise if the directory doesn't exist.
122
123			:param str name: the name of the directory to delete, passed to :func:`shutil.rmtree`.
124			"""
125			def __init__(self, name, label=DEFAULT, args, *kwds):
126			"""
127			@todoc
128			"""
129			Action.__init__(self, "rm -r {}".format(name) if label is DEFAULT else label, args, *kwds)
130			self.__name = name
131
132			def do_execute(self, dependency_statuses):
133			try:
134			shutil.rmtree(self.__name)
135			except OSError as e:
136			if e.errno != errno.ENOENT:
137			raise
138
139
140			class CopyFile(Action): # Not doctested: could be
141			"""
142			A stock action that copies a file. Arguments are passed to :func:`shutil.copy`.
143
144			:param str src: the file to copy
145			:param str dst: the destination
146			"""
147			def __init__(self, src, dst, label=DEFAULT, args, *kwds):
148			"""
149			@todoc
150			"""
151			Action.__init__(self, "cp {} {}".format(src, dst) if label is DEFAULT else label, args, *kwds)
152			self.__src = src
153			self.__dst = dst
154
155			def do_execute(self, dependency_statuses):
156			shutil.copy(self.__src, self.__dst)
157
158
159			class TouchFile(Action): # Not doctested: could be
160			"""
161			A stock action that touches a file.
162			If the file already exists, its modification time will be modified.
163			Else, it will be created, empty.
164
165			Note that the containing directory must exist.
166			You might want to ensure that by adding a :class:`CreateDirectory` as a dependency.
167
168			:param str name: the name of the file to touch. Passed to :func:`open` and/or :func:`os.utime`.
169			"""
170
171			def __init__(self, name, label=DEFAULT, args, *kwds):
172			"""
173			@todoc
174			"""
175			Action.__init__(self, "touch {}".format(name) if label is DEFAULT else label, args, *kwds)
176			self.__name = name
177
178			def do_execute(self, dependency_statuses):
179			open(self.__name, "ab").close() # Create the file if needed
180			os.utime(self.__name, None) # Actually change its time
181
182
183			class Sleep(Action):
184			"""
185			A stock action that sleeps for a certain duration.
186
187			:param float secs: seconds to sleep, passed to :func:`time.sleep`.
188			"""
189			def __init__(self, secs, label=DEFAULT, args, *kwds):
190			"""
191			@todoc
192			"""
193			Action.__init__(self, "sleep {}".format(secs) if label is DEFAULT else label, args, *kwds)
194			self.__secs = secs
195
196			def do_execute(self, dependency_statuses):
197			time.sleep(self.__secs)
198

jacquev6 / ActionTree

Push — master ( 31c7ae...e8f49f )

DeleteDirectory.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

DeleteDirectory.init() A