Code Duplication - drobotun/virustotalapi3 - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 198-198 lines in 2 locations

vtapi3/vtapi3domains.py 1 location


from .vtapi3base import VirusTotalAPI
from .vtapi3error import VirusTotalAPIError

class VirusTotalAPIDomains(VirusTotalAPI):
    """The retrieving information about any domain from the VirusTotal database methods are defined
       in the class.

       Methods:
          get_report(): Retrieve information about an Internet domain.
          get_comments(): Retrieve comments for an Internet domain.
          put_comments(): Add a comment to an Internet domain.
          get_relationship(): Retrieve objects related to an Internet domain.
          get_votes(): Retrieve votes for a hostname or domain.
          put_votes(): Add a vote for a hostname or domain.
    """

    def get_report(self, domain):
        """Retrieve information about an Internet domain.

        Args:
           domain: Domain name (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/domains/' + domain
        try:
            response = requests.get(api_url, headers=self.headers,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_comments(self, domain, limit=10, cursor='""'):
        """Retrieve comments for an Internet domain.

        Args:
           domain: Domain name (str).
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/domains/' + domain + '/comments'
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def put_comments(self, domain, text):
        """Add a comment to an Internet domain.

        Args:
           domain: Domain name (str).
           text: Text of the comment (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        comments = {"data": {'type': 'comment', 'attributes': {'text': text}}}
        api_url = self.base_url + '/domains/' + domain + '/comments'
        try:
            response = requests.post(api_url, headers=self.headers, json=comments,
                                     timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_relationship(self, domain, relationship='/resolutions', limit=10, cursor='""'):
        """Retrieve objects related to an Internet domain.

        Args:
           domain: Domain name (str).
           relationship: Relationship name (str). The default value is '/resolutions'.
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/domains/' + domain + relationship
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_votes(self, domain, limit=10, cursor='""'):
        """Retrieve votes for a hostname or domain.

        Args:
           domain: Domain name (str).
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/domains/' + domain + '/votes'
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def put_votes(self, domain, malicious=False):
        """Add a vote for a hostname or domain.

        Args:
           domain: Domain name (str).
           malicious: Determines a malicious (True) or harmless (False) domain (bool).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        if malicious:
            verdict = 'malicious'
        else:
            verdict = 'harmless'
        votes = {'data': {'type': 'vote', 'attributes': {'verdict': verdict}}}
        api_url = self.base_url + '/domains/' + domain + '/votes'
        try:
            response = requests.post(api_url, headers=self.headers, json=votes,
                                     timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content


vtapi3/vtapi3ipaddresses.py 1 location


from .vtapi3base import VirusTotalAPI
from .vtapi3error import VirusTotalAPIError

class VirusTotalAPIIPAddresses(VirusTotalAPI):
    """The retrieving information about any IP addresses from the VirusTotal database methods are
       defined in the class.

       Methods:
          get_report(): Retrieve information about an IP address.
          get_comments(): Retrieve comments for an IP address.
          put_comments(): Add a comment to an IP address.
          get_relationship(): Retrieve objects related to an IP address.
          get_votes(): Retrieve votes for an IP address.
          put_votes(): Add a vote for an IP address.
    """

    def get_report(self, ip_address):
        """Retrieve information about an IP address.

        Args:
           ip_address: IP address (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        api_url = self.base_url + '/ip_addresses/' + ip_address
        try:
            response = requests.get(api_url, headers=self.headers,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_comments(self, ip_address, limit=10, cursor='""'):
        """Retrieve comments for an IP address.

        Args:
           ip_address: IP address (str).
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/ip_addresses/' + ip_address + '/comments'
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def put_comments(self, ip_address, text):
        """Add a comment to an IP address.

        Args:
           ip_address: IP address (str).
           text: Text of the comment (str).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        comments = {"data": {'type': 'comment', 'attributes': {'text': text}}}
        api_url = self.base_url + '/ip_addresses/' + ip_address + '/comments'
        try:
            response = requests.post(api_url, headers=self.headers, json=comments,
                                     timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_relationship(self, ip_address, relationship='/resolutions', limit=10, cursor='""'):
        """Retrieve objects related to an IP address.

        Args:
           ip_address: IP address (str).
           relationship: Relationship name (str). The default value is '/resolutions'.
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/ip_addresses/' + ip_address + relationship
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def get_votes(self, ip_address, limit=10, cursor='""'):
        """Retrieve votes for an IP address.

        Args:
           domain: Domain name (str).
           limit: Maximum number of comments to retrieve (int). The default value is 10.
           cursor: Continuation cursor (str). The default value is ''.

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        query_string = {'limit': str(limit), 'cursor': cursor}
        api_url = self.base_url + '/ip_addresses/' + ip_address + '/votes'
        try:
            response = requests.get(api_url, headers=self.headers, params=query_string,
                                    timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content

    def put_votes(self, ip_address, malicious=False):
        """Add a vote for an IP address.

        Args:
           domain: IP address (str).
           malicious: Determines a malicious (True) or harmless (False) domain (bool).

        Return:
           The response from the server as a byte sequence.

        Exception
           VirusTotalAPIError(Connection error): In case of server connection errors.
           VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
        """
        self._last_http_error = None
        self._last_result = None
        if malicious:
            verdict = 'malicious'
        else:
            verdict = 'harmless'
        votes = {'data': {'type': 'vote', 'attributes': {'verdict': verdict}}}
        api_url = self.base_url + '/ip_addresses/' + ip_address + '/votes'
        try:
            response = requests.post(api_url, headers=self.headers, json=votes,
                                     timeout=self.timeout, proxies=self.proxies)
        except requests.exceptions.Timeout:
            raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
        except requests.exceptions.ConnectionError:
            raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
        else:
            self._last_http_error = response.status_code
            self._last_result = response.content
            return response.content


		@@ 15-212 (lines=198) @@
12		from .vtapi3base import VirusTotalAPI
13		from .vtapi3error import VirusTotalAPIError
14
15		class VirusTotalAPIDomains(VirusTotalAPI):
16		"""The retrieving information about any domain from the VirusTotal database methods are defined
17		in the class.
18
19		Methods:
20		get_report(): Retrieve information about an Internet domain.
21		get_comments(): Retrieve comments for an Internet domain.
22		put_comments(): Add a comment to an Internet domain.
23		get_relationship(): Retrieve objects related to an Internet domain.
24		get_votes(): Retrieve votes for a hostname or domain.
25		put_votes(): Add a vote for a hostname or domain.
26		"""
27
28		def get_report(self, domain):
29		"""Retrieve information about an Internet domain.
30
31		Args:
32		domain: Domain name (str).
33
34		Return:
35		The response from the server as a byte sequence.
36
37		Exception
38		VirusTotalAPIError(Connection error): In case of server connection errors.
39		VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
40		"""
41		self._last_http_error = None
42		self._last_result = None
43		api_url = self.base_url + '/domains/' + domain
44		try:
45		response = requests.get(api_url, headers=self.headers,
46		timeout=self.timeout, proxies=self.proxies)
47		except requests.exceptions.Timeout:
48		raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
49		except requests.exceptions.ConnectionError:
50		raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
51		else:
52		self._last_http_error = response.status_code
53		self._last_result = response.content
54		return response.content
55
56		def get_comments(self, domain, limit=10, cursor='""'):
57		"""Retrieve comments for an Internet domain.
58
59		Args:
60		domain: Domain name (str).
61		limit: Maximum number of comments to retrieve (int). The default value is 10.
62		cursor: Continuation cursor (str). The default value is ''.
63
64		Return:
65		The response from the server as a byte sequence.
66
67		Exception
68		VirusTotalAPIError(Connection error): In case of server connection errors.
69		VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
70		"""
71		self._last_http_error = None
72		self._last_result = None
73		query_string = {'limit': str(limit), 'cursor': cursor}
74		api_url = self.base_url + '/domains/' + domain + '/comments'
75		try:
76		response = requests.get(api_url, headers=self.headers, params=query_string,
77		timeout=self.timeout, proxies=self.proxies)
78		except requests.exceptions.Timeout:
79		raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
80		except requests.exceptions.ConnectionError:
81		raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
82		else:
83		self._last_http_error = response.status_code
84		self._last_result = response.content
85		return response.content
86
87		def put_comments(self, domain, text):
88		"""Add a comment to an Internet domain.
89
90		Args:
91		domain: Domain name (str).
92		text: Text of the comment (str).
93
94		Return:
95		The response from the server as a byte sequence.
96
97		Exception
98		VirusTotalAPIError(Connection error): In case of server connection errors.
99		VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
100		"""
101		self._last_http_error = None
102		self._last_result = None
103		comments = {"data": {'type': 'comment', 'attributes': {'text': text}}}
104		api_url = self.base_url + '/domains/' + domain + '/comments'
105		try:
106		response = requests.post(api_url, headers=self.headers, json=comments,
107		timeout=self.timeout, proxies=self.proxies)
108		except requests.exceptions.Timeout:
109		raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
110		except requests.exceptions.ConnectionError:
111		raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
112		else:
113		self._last_http_error = response.status_code
114		self._last_result = response.content
115		return response.content
116
117		def get_relationship(self, domain, relationship='/resolutions', limit=10, cursor='""'):
118		"""Retrieve objects related to an Internet domain.
119
120		Args:
121		domain: Domain name (str).
122		relationship: Relationship name (str). The default value is '/resolutions'.
123		limit: Maximum number of comments to retrieve (int). The default value is 10.
124		cursor: Continuation cursor (str). The default value is ''.
125
126		Return:
127		The response from the server as a byte sequence.
128
129		Exception
130		VirusTotalAPIError(Connection error): In case of server connection errors.
131		VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
132		"""
133		self._last_http_error = None
134		self._last_result = None
135		query_string = {'limit': str(limit), 'cursor': cursor}
136		api_url = self.base_url + '/domains/' + domain + relationship
137		try:
138		response = requests.get(api_url, headers=self.headers, params=query_string,
139		timeout=self.timeout, proxies=self.proxies)
140		except requests.exceptions.Timeout:
141		raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
142		except requests.exceptions.ConnectionError:
143		raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
144		else:
145		self._last_http_error = response.status_code
146		self._last_result = response.content
147		return response.content
148
149		def get_votes(self, domain, limit=10, cursor='""'):
150		"""Retrieve votes for a hostname or domain.
151
152		Args:
153		domain: Domain name (str).
154		limit: Maximum number of comments to retrieve (int). The default value is 10.
155		cursor: Continuation cursor (str). The default value is ''.
156
157		Return:
158		The response from the server as a byte sequence.
159
160		Exception
161		VirusTotalAPIError(Connection error): In case of server connection errors.
162		VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
163		"""
164		self._last_http_error = None
165		self._last_result = None
166		query_string = {'limit': str(limit), 'cursor': cursor}
167		api_url = self.base_url + '/domains/' + domain + '/votes'
168		try:
169		response = requests.get(api_url, headers=self.headers, params=query_string,
170		timeout=self.timeout, proxies=self.proxies)
171		except requests.exceptions.Timeout:
172		raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
173		except requests.exceptions.ConnectionError:
174		raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
175		else:
176		self._last_http_error = response.status_code
177		self._last_result = response.content
178		return response.content
179
180		def put_votes(self, domain, malicious=False):
181		"""Add a vote for a hostname or domain.
182
183		Args:
184		domain: Domain name (str).
185		malicious: Determines a malicious (True) or harmless (False) domain (bool).
186
187		Return:
188		The response from the server as a byte sequence.
189
190		Exception
191		VirusTotalAPIError(Connection error): In case of server connection errors.
192		VirusTotalAPIError(Timeout error): If the response timeout from the server is exceeded.
193		"""
194		self._last_http_error = None
195		self._last_result = None
196		if malicious:
197		verdict = 'malicious'
198		else:
199		verdict = 'harmless'
200		votes = {'data': {'type': 'vote', 'attributes': {'verdict': verdict}}}
201		api_url = self.base_url + '/domains/' + domain + '/votes'
202		try:
203		response = requests.post(api_url, headers=self.headers, json=votes,
204		timeout=self.timeout, proxies=self.proxies)
205		except requests.exceptions.Timeout:
206		raise VirusTotalAPIError('Timeout error', errno.ETIMEDOUT)
207		except requests.exceptions.ConnectionError:
208		raise VirusTotalAPIError('Connection error', errno.ECONNABORTED)
209		else:
210		self._last_http_error = response.status_code
211		self._last_result = response.content
212		return response.content
213

drobotun / virustotalapi3

Code Duplication Length = 198-198 lines in 2 locations

vtapi3/vtapi3domains.py 1 location

vtapi3/vtapi3ipaddresses.py 1 location