<s> import logging <EOL> from urllib . parse import urlencode , urljoin <EOL> import requests <EOL> from django . conf import settings <EOL> from django . utils . translation import ugettext_lazy as _ <EOL> from ralph . dns . forms import RecordType <EOL> from ralph . helpers import cache <EOL> logger = logging . getLogger ( __name__ ) <EOL> class DNSaaS : <EOL> def __init__ ( self , headers = None ) : <EOL> self . session = requests . Session ( ) <EOL> _headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' . format ( settings . DNSAAS_TOKEN ) <EOL> } <EOL> if headers is not None : <EOL> _headers . update ( headers ) <EOL> self . session . headers . update ( _headers ) <EOL> def get_api_result ( self , url ) : <EOL> """<STR_LIT>""" <EOL> request = self . session . get ( url ) <EOL> json_data = request . json ( ) <EOL> api_results = json_data . get ( '<STR_LIT>' , [ ] ) <EOL> if json_data . get ( '<STR_LIT>' , None ) : <EOL> _api_results = self . get_api_result ( json_data [ '<STR_LIT>' ] ) <EOL> api_results . extend ( _api_results ) <EOL> return api_results <EOL> def get_dns_records ( self , ipaddresses ) : <EOL> """<STR_LIT>""" <EOL> dns_records = [ ] <EOL> ipaddresses = [ ( '<STR_LIT>' , i ) for i in ipaddresses ] <EOL> url = urljoin ( <EOL> settings . DNSAAS_URL , <EOL> '<STR_LIT>' . format ( <EOL> urlencode ( [ <EOL> ( '<STR_LIT>' , <NUM_LIT:100> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> ] + ipaddresses ) <EOL> ) <EOL> ) <EOL> api_results = self . get_api_result ( url ) <EOL> ptrs = set ( [ i [ '<STR_LIT:content>' ] for i in api_results if i [ '<STR_LIT:type>' ] == '<STR_LIT>' ] ) <EOL> for item in api_results : <EOL> if item [ '<STR_LIT:type>' ] in { '<STR_LIT:A>' , '<STR_LIT>' , '<STR_LIT>' } : <EOL> dns_records . append ( { <EOL> '<STR_LIT>' : item [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT:name>' : item [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT:type>' : RecordType . from_name ( item [ '<STR_LIT:type>' ] . lower ( ) ) . id , <EOL> '<STR_LIT:content>' : item [ '<STR_LIT:content>' ] , <EOL> '<STR_LIT>' : item [ '<STR_LIT:name>' ] in ptrs and item [ '<STR_LIT:type>' ] == '<STR_LIT:A>' , <EOL> '<STR_LIT>' : settings . DNSAAS_OWNER <EOL> } ) <EOL> return sorted ( dns_records , key = lambda x : x [ '<STR_LIT:type>' ] ) <EOL> def update_dns_record ( self , record ) : <EOL> """<STR_LIT>""" <EOL> url = urljoin ( <EOL> settings . DNSAAS_URL , '<STR_LIT>' . format ( record [ '<STR_LIT>' ] ) <EOL> ) <EOL> data = { <EOL> '<STR_LIT:name>' : record [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT:type>' : RecordType . raw_from_id ( int ( record [ '<STR_LIT:type>' ] ) ) , <EOL> '<STR_LIT:content>' : record [ '<STR_LIT:content>' ] , <EOL> '<STR_LIT>' : ( <EOL> settings . DNSAAS_AUTO_PTR_ALWAYS if record [ '<STR_LIT>' ] and <EOL> record [ '<STR_LIT:type>' ] == str ( RecordType . a . id ) <EOL> else settings . DNSAAS_AUTO_PTR_NEVER <EOL> ) , <EOL> '<STR_LIT>' : settings . DNSAAS_OWNER <EOL> } <EOL> request = self . session . patch ( url , data = data ) <EOL> if request . status_code == <NUM_LIT> : <EOL> return { <EOL> '<STR_LIT>' : [ _ ( '<STR_LIT>' ) ] <EOL> } <EOL> elif request . status_code != <NUM_LIT:200> : <EOL> return request . json ( ) <EOL> @ cache ( skip_first = True ) <EOL> def get_domain ( self , domain_name ) : <EOL> """<STR_LIT>""" <EOL> url = urljoin ( <EOL> settings . DNSAAS_URL , '<STR_LIT>' . format ( <EOL> urlencode ( [ ( '<STR_LIT:name>' , domain_name ) ] ) <EOL> ) <EOL> ) <EOL> result = self . get_api_result ( url ) <EOL> if result : <EOL> return result [ <NUM_LIT:0> ] [ '<STR_LIT:url>' ] <EOL> def create_dns_record ( self , record ) : <EOL> """<STR_LIT>""" <EOL> url = urljoin ( settings . DNSAAS_URL , '<STR_LIT>' ) <EOL> domain_name = record [ '<STR_LIT:name>' ] . split ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> domain = self . get_domain ( domain_name [ - <NUM_LIT:1> ] ) <EOL> if not domain : <EOL> logger . error ( <EOL> '<STR_LIT>' . format ( record ) <EOL> ) <EOL> return { '<STR_LIT:name>' : [ _ ( '<STR_LIT>' ) ] } <EOL> data = { <EOL> '<STR_LIT:name>' : record [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT:type>' : RecordType . raw_from_id ( int ( record [ '<STR_LIT:type>' ] ) ) , <EOL> '<STR_LIT:content>' : record [ '<STR_LIT:content>' ] , <EOL> '<STR_LIT>' : ( <EOL> settings . DNSAAS_AUTO_PTR_ALWAYS if record [ '<STR_LIT>' ] and <EOL> record [ '<STR_LIT:type>' ] == RecordType . a . id <EOL> else settings . DNSAAS_AUTO_PTR_NEVER <EOL> ) , <EOL> '<STR_LIT>' : domain , <EOL> '<STR_LIT>' : settings . DNSAAS_OWNER <EOL> } <EOL> request = self . session . post ( url , data = data ) <EOL> if request . status_code == <NUM_LIT> : <EOL> return { <EOL> '<STR_LIT>' : [ _ ( '<STR_LIT>' ) ] <EOL> } <EOL> elif request . status_code != <NUM_LIT> : <EOL> return request . json ( ) <EOL> def delete_dns_record ( self , record_id ) : <EOL> """<STR_LIT>""" <EOL> url = urljoin ( <EOL> settings . DNSAAS_URL , '<STR_LIT>' . format ( record_id ) <EOL> ) <EOL> request = self . session . delete ( url ) <EOL> if request . status_code == <NUM_LIT> : <EOL> return { <EOL> '<STR_LIT>' : [ _ ( '<STR_LIT>' ) ] <EOL> } <EOL> elif request . status_code != <NUM_LIT> : <EOL> return request . json ( ) </s>
