<s> import pycurl <EOL> import os <EOL> import shutil <EOL> import threading <EOL> import lxml . etree as etree <EOL> from io import BytesIO <EOL> from re import sub <EOL> from webdav . connection import * <EOL> from webdav . exceptions import * <EOL> from webdav . urn import Urn <EOL> try : <EOL> from urllib . parse import unquote <EOL> except ImportError : <EOL> from urllib import unquote <EOL> __version__ = "<STR_LIT>" <EOL> def listdir ( directory ) : <EOL> file_names = list ( ) <EOL> for filename in os . listdir ( directory ) : <EOL> file_path = os . path . join ( directory , filename ) <EOL> if os . path . isdir ( file_path ) : <EOL> filename = "<STR_LIT>" . format ( filename = filename , separate = os . path . sep ) <EOL> file_names . append ( filename ) <EOL> return file_names <EOL> def add_options ( request , options ) : <EOL> for ( key , value ) in options . items ( ) : <EOL> if value is None : <EOL> continue <EOL> try : <EOL> request . setopt ( pycurl . __dict__ [ key ] , value ) <EOL> except TypeError : <EOL> raise OptionNotValid ( key , value ) <EOL> except pycurl . error : <EOL> raise OptionNotValid ( key , value ) <EOL> def get_options ( type , from_options ) : <EOL> _options = dict ( ) <EOL> for key in type . keys : <EOL> key_with_prefix = "<STR_LIT>" . format ( prefix = type . prefix , key = key ) <EOL> if key not in from_options and key_with_prefix not in from_options : <EOL> _options [ key ] = "<STR_LIT>" <EOL> elif key in from_options : <EOL> _options [ key ] = from_options . get ( key ) <EOL> else : <EOL> _options [ key ] = from_options . get ( key_with_prefix ) <EOL> return _options <EOL> class Client ( object ) : <EOL> root = '<STR_LIT:/>' <EOL> large_size = <NUM_LIT:2> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> <EOL> http_header = { <EOL> '<STR_LIT:list>' : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> '<STR_LIT>' : [ "<STR_LIT>" ] , <EOL> '<STR_LIT>' : [ "<STR_LIT>" ] , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> '<STR_LIT:info>' : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> } <EOL> def get_header ( self , method ) : <EOL> if method in Client . http_header : <EOL> try : <EOL> header = Client . http_header [ method ] . copy ( ) <EOL> except AttributeError : <EOL> header = Client . http_header [ method ] [ : ] <EOL> else : <EOL> header = list ( ) <EOL> if self . webdav . token : <EOL> webdav_token = "<STR_LIT>" . format ( token = self . webdav . token ) <EOL> header . append ( webdav_token ) <EOL> return header <EOL> requests = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:list>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:info>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> } <EOL> meta_xmlns = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> } <EOL> def __init__ ( self , options ) : <EOL> webdav_options = get_options ( type = WebDAVSettings , from_options = options ) <EOL> proxy_options = get_options ( type = ProxySettings , from_options = options ) <EOL> self . webdav = WebDAVSettings ( webdav_options ) <EOL> self . proxy = ProxySettings ( proxy_options ) <EOL> pycurl . global_init ( pycurl . GLOBAL_DEFAULT ) <EOL> self . default_options = { } <EOL> def __del__ ( self ) : <EOL> pycurl . global_cleanup ( ) <EOL> def valid ( self ) : <EOL> return True if self . webdav . valid ( ) and self . proxy . valid ( ) else False <EOL> def Request ( self , options = None ) : <EOL> curl = pycurl . Curl ( ) <EOL> self . default_options . update ( { <EOL> '<STR_LIT>' : self . webdav . hostname , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : pycurl . SSLVERSION_TLSv1 , <EOL> } ) <EOL> if not self . webdav . token : <EOL> server_token = '<STR_LIT>' . format ( login = self . webdav . login , password = self . webdav . password ) <EOL> self . default_options . update ( { <EOL> '<STR_LIT>' : server_token , <EOL> } ) <EOL> if self . proxy . valid ( ) : <EOL> if self . proxy . hostname : <EOL> self . default_options [ '<STR_LIT>' ] = self . proxy . hostname <EOL> if self . proxy . login : <EOL> if not self . proxy . password : <EOL> self . default_options [ '<STR_LIT>' ] = self . proxy . login <EOL> else : <EOL> proxy_token = '<STR_LIT>' . format ( login = self . proxy . login , password = self . proxy . password ) <EOL> self . default_options [ '<STR_LIT>' ] = proxy_token <EOL> if self . webdav . cert_path : <EOL> self . default_options [ '<STR_LIT>' ] = self . webdav . cert_path <EOL> if self . webdav . key_path : <EOL> self . default_options [ '<STR_LIT>' ] = self . webdav . key_path <EOL> if self . default_options : <EOL> add_options ( curl , self . default_options ) <EOL> if options : <EOL> add_options ( curl , options ) <EOL> return curl <EOL> def list ( self , remote_path = root ) : <EOL> def parse ( response ) : <EOL> try : <EOL> response_str = response . getvalue ( ) <EOL> tree = etree . fromstring ( response_str ) <EOL> hrees = [ unquote ( hree . text ) for hree in tree . findall ( "<STR_LIT>" ) ] <EOL> return [ Urn ( hree ) for hree in hrees ] <EOL> except etree . XMLSyntaxError : <EOL> return list ( ) <EOL> try : <EOL> directory_urn = Urn ( remote_path , directory = True ) <EOL> if directory_urn . path ( ) != Client . root : <EOL> if not self . check ( directory_urn . path ( ) ) : <EOL> raise RemoteResourceNotFound ( directory_urn . path ( ) ) <EOL> response = BytesIO ( ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : directory_urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT:list>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT:list>' ) , <EOL> '<STR_LIT>' : response , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> urns = parse ( response ) <EOL> path = "<STR_LIT>" . format ( root = self . webdav . root , path = directory_urn . path ( ) ) <EOL> return [ urn . filename ( ) for urn in urns if urn . path ( ) != path and urn . path ( ) != path [ : - <NUM_LIT:1> ] ] <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def free ( self ) : <EOL> def parse ( response ) : <EOL> try : <EOL> response_str = response . getvalue ( ) <EOL> tree = etree . fromstring ( response_str ) <EOL> node = tree . find ( '<STR_LIT>' ) <EOL> if node is not None : <EOL> return int ( node . text ) <EOL> else : <EOL> raise MethodNotSupported ( name = '<STR_LIT>' , server = self . webdav . hostname ) <EOL> except TypeError : <EOL> raise MethodNotSupported ( name = '<STR_LIT>' , server = self . webdav . hostname ) <EOL> except etree . XMLSyntaxError : <EOL> return str ( ) <EOL> def data ( ) : <EOL> root = etree . Element ( "<STR_LIT>" , xmlns = "<STR_LIT>" ) <EOL> prop = etree . SubElement ( root , "<STR_LIT>" ) <EOL> etree . SubElement ( prop , "<STR_LIT>" ) <EOL> etree . SubElement ( prop , "<STR_LIT>" ) <EOL> tree = etree . ElementTree ( root ) <EOL> buff = BytesIO ( ) <EOL> tree . write ( buff ) <EOL> return buff . getvalue ( ) <EOL> try : <EOL> response = BytesIO ( ) <EOL> options = { <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : data ( ) , <EOL> '<STR_LIT>' : response , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> return parse ( response ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def check ( self , remote_path = root ) : <EOL> def parse ( response , path ) : <EOL> try : <EOL> response_str = response . getvalue ( ) <EOL> tree = etree . fromstring ( response_str ) <EOL> resps = tree . findall ( "<STR_LIT>" ) <EOL> for resp in resps : <EOL> href = resp . findtext ( "<STR_LIT>" ) <EOL> urn = unquote ( href ) <EOL> if path . endswith ( Urn . separate ) : <EOL> if path == urn or path [ : - <NUM_LIT:1> ] == urn : <EOL> return True <EOL> else : <EOL> if path == urn : <EOL> return True <EOL> return False <EOL> except etree . XMLSyntaxError : <EOL> raise MethodNotSupported ( name = "<STR_LIT>" , server = self . webdav . hostname ) <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> parent_urn = Urn ( urn . parent ( ) ) <EOL> response = BytesIO ( ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : parent_urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT:info>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT:info>' ) , <EOL> '<STR_LIT>' : response , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> path = "<STR_LIT>" . format ( root = self . webdav . root , path = urn . path ( ) ) <EOL> return parse ( response , path ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def mkdir ( self , remote_path ) : <EOL> try : <EOL> directory_urn = Urn ( remote_path , directory = True ) <EOL> if not self . check ( directory_urn . parent ( ) ) : <EOL> raise RemoteParentNotFound ( directory_urn . path ( ) ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : directory_urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def download_to ( self , buff , remote_path ) : <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> if self . is_dir ( urn . path ( ) ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = remote_path ) <EOL> if not self . check ( urn . path ( ) ) : <EOL> raise RemoteResourceNotFound ( urn . path ( ) ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : buff . write , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def download ( self , remote_path , local_path , progress = None ) : <EOL> urn = Urn ( remote_path ) <EOL> if self . is_dir ( urn . path ( ) ) : <EOL> self . download_directory ( local_path = local_path , remote_path = remote_path , progress = progress ) <EOL> else : <EOL> self . download_file ( local_path = local_path , remote_path = remote_path , progress = progress ) <EOL> def download_directory ( self , remote_path , local_path , progress = None ) : <EOL> urn = Urn ( remote_path , directory = True ) <EOL> if not self . is_dir ( urn . path ( ) ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = remote_path ) <EOL> if os . path . exists ( local_path ) : <EOL> shutil . rmtree ( local_path ) <EOL> os . makedirs ( local_path ) <EOL> for resource_name in self . list ( urn . path ( ) ) : <EOL> _remote_path = "<STR_LIT>" . format ( parent = urn . path ( ) , name = resource_name ) <EOL> _local_path = os . path . join ( local_path , resource_name ) <EOL> self . download ( local_path = _local_path , remote_path = _remote_path , progress = progress ) <EOL> def download_file ( self , remote_path , local_path , progress = None ) : <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> if self . is_dir ( urn . path ( ) ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = remote_path ) <EOL> if os . path . isdir ( local_path ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = local_path ) <EOL> if not self . check ( urn . path ( ) ) : <EOL> raise RemoteResourceNotFound ( urn . path ( ) ) <EOL> with open ( local_path , '<STR_LIT:wb>' ) as local_file : <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : local_file , <EOL> '<STR_LIT>' : <NUM_LIT:0> if progress else <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> if progress : <EOL> options [ "<STR_LIT>" ] = progress <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def download_sync ( self , remote_path , local_path , callback = None ) : <EOL> self . download ( local_path = local_path , remote_path = remote_path ) <EOL> if callback : <EOL> callback ( ) <EOL> def download_async ( self , remote_path , local_path , callback = None ) : <EOL> target = ( lambda : self . download_sync ( local_path = local_path , remote_path = remote_path , callback = callback ) ) <EOL> threading . Thread ( target = target ) . start ( ) <EOL> def upload_from ( self , buff , remote_path ) : <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> if urn . is_dir ( ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = remote_path ) <EOL> if not self . check ( urn . parent ( ) ) : <EOL> raise RemoteParentNotFound ( urn . path ( ) ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : buff . read , <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> code = request . getinfo ( pycurl . HTTP_CODE ) <EOL> if code == "<STR_LIT>" : <EOL> raise NotEnoughSpace ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def upload ( self , remote_path , local_path , progress = None ) : <EOL> if os . path . isdir ( local_path ) : <EOL> self . upload_directory ( local_path = local_path , remote_path = remote_path , progress = progress ) <EOL> else : <EOL> self . upload_file ( local_path = local_path , remote_path = remote_path , progress = progress ) <EOL> def upload_directory ( self , remote_path , local_path , progress = None ) : <EOL> urn = Urn ( remote_path , directory = True ) <EOL> if not urn . is_dir ( ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = remote_path ) <EOL> if not os . path . isdir ( local_path ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = local_path ) <EOL> if not os . path . exists ( local_path ) : <EOL> raise LocalResourceNotFound ( local_path ) <EOL> if self . check ( urn . path ( ) ) : <EOL> self . clean ( urn . path ( ) ) <EOL> self . mkdir ( remote_path ) <EOL> for resource_name in listdir ( local_path ) : <EOL> _remote_path = "<STR_LIT>" . format ( parent = urn . path ( ) , name = resource_name ) <EOL> _local_path = os . path . join ( local_path , resource_name ) <EOL> self . upload ( local_path = _local_path , remote_path = _remote_path , progress = progress ) <EOL> def upload_file ( self , remote_path , local_path , progress = None ) : <EOL> try : <EOL> if not os . path . exists ( local_path ) : <EOL> raise LocalResourceNotFound ( local_path ) <EOL> urn = Urn ( remote_path ) <EOL> if urn . is_dir ( ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = remote_path ) <EOL> if os . path . isdir ( local_path ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = local_path ) <EOL> if not os . path . exists ( local_path ) : <EOL> raise LocalResourceNotFound ( local_path ) <EOL> if not self . check ( urn . parent ( ) ) : <EOL> raise RemoteParentNotFound ( urn . path ( ) ) <EOL> with open ( local_path , "<STR_LIT:rb>" ) as local_file : <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : local_file . read , <EOL> '<STR_LIT>' : <NUM_LIT:0> if progress else <NUM_LIT:1> <EOL> } <EOL> if progress : <EOL> options [ "<STR_LIT>" ] = progress <EOL> file_size = os . path . getsize ( local_path ) <EOL> if file_size > self . large_size : <EOL> options [ '<STR_LIT>' ] = file_size <EOL> else : <EOL> options [ '<STR_LIT>' ] = file_size <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> code = request . getinfo ( pycurl . HTTP_CODE ) <EOL> if code == "<STR_LIT>" : <EOL> raise NotEnoughSpace ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def upload_sync ( self , remote_path , local_path , callback = None ) : <EOL> self . upload ( local_path = local_path , remote_path = remote_path ) <EOL> if callback : <EOL> callback ( ) <EOL> def upload_async ( self , remote_path , local_path , callback = None ) : <EOL> target = ( lambda : self . upload_sync ( local_path = local_path , remote_path = remote_path , callback = callback ) ) <EOL> threading . Thread ( target = target ) . start ( ) <EOL> def copy ( self , remote_path_from , remote_path_to ) : <EOL> def header ( remote_path_to ) : <EOL> path = Urn ( remote_path_to ) . path ( ) <EOL> destination = "<STR_LIT>" . format ( root = self . webdav . root , path = path ) <EOL> header_item = "<STR_LIT>" . format ( destination = destination ) <EOL> header = self . get_header ( '<STR_LIT>' ) <EOL> header . append ( header_item ) <EOL> return header <EOL> try : <EOL> urn_from = Urn ( remote_path_from ) <EOL> if not self . check ( urn_from . path ( ) ) : <EOL> raise RemoteResourceNotFound ( urn_from . path ( ) ) <EOL> urn_to = Urn ( remote_path_to ) <EOL> if not self . check ( urn_to . parent ( ) ) : <EOL> raise RemoteParentNotFound ( urn_to . path ( ) ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn_from . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : header ( remote_path_to ) <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def move ( self , remote_path_from , remote_path_to ) : <EOL> def header ( remote_path_to ) : <EOL> path = Urn ( remote_path_to ) . path ( ) <EOL> destination = "<STR_LIT>" . format ( root = self . webdav . root , path = path ) <EOL> header_item = "<STR_LIT>" . format ( destination = destination ) <EOL> header = self . get_header ( '<STR_LIT>' ) <EOL> header . append ( header_item ) <EOL> return header <EOL> try : <EOL> urn_from = Urn ( remote_path_from ) <EOL> if not self . check ( urn_from . path ( ) ) : <EOL> raise RemoteResourceNotFound ( urn_from . path ( ) ) <EOL> urn_to = Urn ( remote_path_to ) <EOL> if not self . check ( urn_to . parent ( ) ) : <EOL> raise RemoteParentNotFound ( urn_to . path ( ) ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn_from . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : header ( remote_path_to ) <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def clean ( self , remote_path ) : <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def publish ( self , remote_path ) : <EOL> def parse ( response ) : <EOL> try : <EOL> response_str = response . getvalue ( ) <EOL> tree = etree . fromstring ( response_str ) <EOL> result = tree . xpath ( "<STR_LIT>" ) <EOL> public_url = result [ <NUM_LIT:0> ] <EOL> return public_url . text <EOL> except IndexError : <EOL> raise MethodNotSupported ( name = "<STR_LIT>" , server = self . webdav . hostname ) <EOL> except etree . XMLSyntaxError : <EOL> return "<STR_LIT>" <EOL> def data ( for_server ) : <EOL> root_node = etree . Element ( "<STR_LIT>" , xmlns = "<STR_LIT>" ) <EOL> set_node = etree . SubElement ( root_node , "<STR_LIT>" ) <EOL> prop_node = etree . SubElement ( set_node , "<STR_LIT>" ) <EOL> xmlns = Client . meta_xmlns . get ( for_server , "<STR_LIT>" ) <EOL> public_url = etree . SubElement ( prop_node , "<STR_LIT>" , xmlns = xmlns ) <EOL> public_url . text = "<STR_LIT:true>" <EOL> tree = etree . ElementTree ( root_node ) <EOL> buff = BytesIO ( ) <EOL> tree . write ( buff ) <EOL> return buff . getvalue ( ) <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> if not self . check ( urn . path ( ) ) : <EOL> raise RemoteResourceNotFound ( urn . path ( ) ) <EOL> response = BytesIO ( ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : data ( for_server = self . webdav . hostname ) , <EOL> '<STR_LIT>' : response , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> return parse ( response ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def unpublish ( self , remote_path ) : <EOL> def data ( for_server ) : <EOL> root = etree . Element ( "<STR_LIT>" , xmlns = "<STR_LIT>" ) <EOL> remove = etree . SubElement ( root , "<STR_LIT>" ) <EOL> prop = etree . SubElement ( remove , "<STR_LIT>" ) <EOL> xmlns = Client . meta_xmlns . get ( for_server , "<STR_LIT>" ) <EOL> etree . SubElement ( prop , "<STR_LIT>" , xmlns = xmlns ) <EOL> tree = etree . ElementTree ( root ) <EOL> buff = BytesIO ( ) <EOL> tree . write ( buff ) <EOL> return buff . getvalue ( ) <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> if not self . check ( urn . path ( ) ) : <EOL> raise RemoteResourceNotFound ( urn . path ( ) ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : data ( for_server = self . webdav . hostname ) <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def info ( self , remote_path ) : <EOL> def parse ( response , path ) : <EOL> try : <EOL> response_str = response . getvalue ( ) <EOL> tree = etree . fromstring ( response_str ) <EOL> find_attributes = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:name>' : "<STR_LIT>" , <EOL> '<STR_LIT:size>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> } <EOL> resps = tree . findall ( "<STR_LIT>" ) <EOL> for resp in resps : <EOL> href = resp . findtext ( "<STR_LIT>" ) <EOL> urn = unquote ( href ) <EOL> if path [ - <NUM_LIT:1> ] == Urn . separate : <EOL> if not path == urn : <EOL> continue <EOL> else : <EOL> path_with_sep = "<STR_LIT>" . format ( path = path , sep = Urn . separate ) <EOL> if not path == urn and not path_with_sep == urn : <EOL> continue <EOL> info = dict ( ) <EOL> for ( name , value ) in find_attributes . items ( ) : <EOL> info [ name ] = resp . findtext ( value ) <EOL> return info <EOL> raise RemoteResourceNotFound ( path ) <EOL> except etree . XMLSyntaxError : <EOL> raise MethodNotSupported ( name = "<STR_LIT:info>" , server = self . webdav . hostname ) <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> response = BytesIO ( ) <EOL> if not self . check ( urn . path ( ) ) and not self . check ( Urn ( remote_path , directory = True ) . path ( ) ) : <EOL> raise RemoteResourceNotFound ( remote_path ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT:info>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT:info>' ) , <EOL> '<STR_LIT>' : response , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> path = "<STR_LIT>" . format ( root = self . webdav . root , path = urn . path ( ) ) <EOL> return parse ( response , path ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def is_dir ( self , remote_path ) : <EOL> def parse ( response , path ) : <EOL> try : <EOL> response_str = response . getvalue ( ) <EOL> tree = etree . fromstring ( response_str ) <EOL> resps = tree . findall ( "<STR_LIT>" ) <EOL> for resp in resps : <EOL> href = resp . findtext ( "<STR_LIT>" ) <EOL> urn = unquote ( href ) <EOL> if path [ - <NUM_LIT:1> ] == Urn . separate : <EOL> if not path == urn : <EOL> continue <EOL> else : <EOL> path_with_sep = "<STR_LIT>" . format ( path = path , sep = Urn . separate ) <EOL> if not path == urn and not path_with_sep == urn : <EOL> continue <EOL> type = resp . find ( "<STR_LIT>" ) <EOL> if type is None : <EOL> raise MethodNotSupported ( name = "<STR_LIT>" , server = self . webdav . hostname ) <EOL> dir_type = type . find ( "<STR_LIT>" ) <EOL> return True if dir_type is not None else False <EOL> raise RemoteResourceNotFound ( path ) <EOL> except etree . XMLSyntaxError : <EOL> raise MethodNotSupported ( name = "<STR_LIT>" , server = self . webdav . hostname ) <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> parent_urn = Urn ( urn . parent ( ) ) <EOL> if not self . check ( urn . path ( ) ) and not self . check ( Urn ( remote_path , directory = True ) . path ( ) ) : <EOL> raise RemoteResourceNotFound ( remote_path ) <EOL> response = BytesIO ( ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : parent_urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT:info>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT:info>' ) , <EOL> '<STR_LIT>' : response , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> path = "<STR_LIT>" . format ( root = self . webdav . root , path = urn . path ( ) ) <EOL> return parse ( response , path ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def resource ( self , remote_path ) : <EOL> urn = Urn ( remote_path ) <EOL> return Resource ( self , urn . path ( ) ) <EOL> def get_property ( self , remote_path , option ) : <EOL> def parse ( response , option ) : <EOL> response_str = response . getvalue ( ) <EOL> tree = etree . fromstring ( response_str ) <EOL> xpath = "<STR_LIT>" . format ( xpath_prefix = "<STR_LIT>" , xpath_exp = option [ '<STR_LIT:name>' ] ) <EOL> return tree . findtext ( xpath ) <EOL> def data ( option ) : <EOL> root = etree . Element ( "<STR_LIT>" , xmlns = "<STR_LIT>" ) <EOL> prop = etree . SubElement ( root , "<STR_LIT>" ) <EOL> etree . SubElement ( prop , option . get ( '<STR_LIT:name>' , "<STR_LIT>" ) , xmlns = option . get ( '<STR_LIT>' , "<STR_LIT>" ) ) <EOL> tree = etree . ElementTree ( root ) <EOL> buff = BytesIO ( ) <EOL> tree . write ( buff ) <EOL> return buff . getvalue ( ) <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> if not self . check ( urn . path ( ) ) : <EOL> raise RemoteResourceNotFound ( urn . path ( ) ) <EOL> response = BytesIO ( ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : data ( option ) , <EOL> '<STR_LIT>' : response , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> return parse ( response , option ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def set_property ( self , remote_path , option ) : <EOL> def data ( option ) : <EOL> root_node = etree . Element ( "<STR_LIT>" , xmlns = "<STR_LIT>" ) <EOL> root_node . set ( '<STR_LIT>' , option . get ( '<STR_LIT>' , "<STR_LIT>" ) ) <EOL> set_node = etree . SubElement ( root_node , "<STR_LIT>" ) <EOL> prop_node = etree . SubElement ( set_node , "<STR_LIT>" ) <EOL> opt_node = etree . SubElement ( prop_node , "<STR_LIT>" . format ( namespace = '<STR_LIT:u>' , name = option [ '<STR_LIT:name>' ] ) ) <EOL> opt_node . text = option . get ( '<STR_LIT:value>' , "<STR_LIT>" ) <EOL> tree = etree . ElementTree ( root_node ) <EOL> buff = BytesIO ( ) <EOL> tree . write ( buff ) <EOL> return buff . getvalue ( ) <EOL> try : <EOL> urn = Urn ( remote_path ) <EOL> if not self . check ( urn . path ( ) ) : <EOL> raise RemoteResourceNotFound ( urn . path ( ) ) <EOL> url = { '<STR_LIT>' : self . webdav . hostname , '<STR_LIT:root>' : self . webdav . root , '<STR_LIT:path>' : urn . quote ( ) } <EOL> options = { <EOL> '<STR_LIT>' : "<STR_LIT>" . format ( ** url ) , <EOL> '<STR_LIT>' : Client . requests [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . get_header ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : data ( option ) <EOL> } <EOL> request = self . Request ( options = options ) <EOL> request . perform ( ) <EOL> request . close ( ) <EOL> except pycurl . error : <EOL> raise NotConnection ( self . webdav . hostname ) <EOL> def push ( self , remote_directory , local_directory ) : <EOL> def prune ( src , exp ) : <EOL> return [ sub ( exp , "<STR_LIT>" , item ) for item in src ] <EOL> urn = Urn ( remote_directory , directory = True ) <EOL> if not self . is_dir ( urn . path ( ) ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = remote_directory ) <EOL> if not os . path . isdir ( local_directory ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = local_directory ) <EOL> if not os . path . exists ( local_directory ) : <EOL> raise LocalResourceNotFound ( local_directory ) <EOL> paths = self . list ( urn . path ( ) ) <EOL> expression = "<STR_LIT>" . format ( begin = "<STR_LIT>" , end = urn . path ( ) ) <EOL> remote_resource_names = prune ( paths , expression ) <EOL> for local_resource_name in listdir ( local_directory ) : <EOL> local_path = os . path . join ( local_directory , local_resource_name ) <EOL> remote_path = "<STR_LIT>" . format ( remote_directory = urn . path ( ) , resource_name = local_resource_name ) <EOL> if os . path . isdir ( local_path ) : <EOL> if not self . check ( remote_path = remote_path ) : <EOL> self . mkdir ( remote_path = remote_path ) <EOL> self . push ( remote_directory = remote_path , local_directory = local_path ) <EOL> else : <EOL> if local_resource_name in remote_resource_names : <EOL> continue <EOL> self . upload_file ( remote_path = remote_path , local_path = local_path ) <EOL> def pull ( self , remote_directory , local_directory ) : <EOL> def prune ( src , exp ) : <EOL> return [ sub ( exp , "<STR_LIT>" , item ) for item in src ] <EOL> urn = Urn ( remote_directory , directory = True ) <EOL> if not self . is_dir ( urn . path ( ) ) : <EOL> raise OptionNotValid ( name = "<STR_LIT>" , value = remote_directory ) <EOL> if not os . path . exists ( local_directory ) : <EOL> raise LocalResourceNotFound ( local_directory ) <EOL> local_resource_names = listdir ( local_directory ) <EOL> paths = self . list ( urn . path ( ) ) <EOL> expression = "<STR_LIT>" . format ( begin = "<STR_LIT>" , end = remote_directory ) <EOL> remote_resource_names = prune ( paths , expression ) <EOL> for remote_resource_name in remote_resource_names : <EOL> local_path = os . path . join ( local_directory , remote_resource_name ) <EOL> remote_path = "<STR_LIT>" . format ( remote_directory = urn . path ( ) , resource_name = remote_resource_name ) <EOL> remote_urn = Urn ( remote_path ) <EOL> if self . is_dir ( remote_urn . path ( ) ) : <EOL> if not os . path . exists ( local_path ) : <EOL> os . mkdir ( local_path ) <EOL> self . pull ( remote_directory = remote_path , local_directory = local_path ) <EOL> else : <EOL> if remote_resource_name in local_resource_names : <EOL> continue <EOL> self . download_file ( remote_path = remote_path , local_path = local_path ) <EOL> def sync ( self , remote_directory , local_directory ) : <EOL> self . pull ( remote_directory = remote_directory , local_directory = local_directory ) <EOL> self . push ( remote_directory = remote_directory , local_directory = local_directory ) <EOL> class Resource ( object ) : <EOL> def __init__ ( self , client , urn ) : <EOL> self . client = client <EOL> self . urn = urn <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" . format ( path = self . urn . path ( ) ) <EOL> def is_dir ( self ) : <EOL> return self . client . is_dir ( self . urn . path ( ) ) <EOL> def rename ( self , new_name ) : <EOL> old_path = self . urn . path ( ) <EOL> parent_path = self . urn . parent ( ) <EOL> new_name = Urn ( new_name ) . filename ( ) <EOL> new_path = "<STR_LIT>" . format ( directory = parent_path , filename = new_name ) <EOL> self . client . move ( remote_path_from = old_path , remote_path_to = new_path ) <EOL> self . urn = Urn ( new_path ) <EOL> def move ( self , remote_path ) : <EOL> new_urn = Urn ( remote_path ) <EOL> self . client . move ( remote_path_from = self . urn . path ( ) , remote_path_to = new_urn . path ( ) ) <EOL> self . urn = new_urn <EOL> def copy ( self , remote_path ) : <EOL> urn = Urn ( remote_path ) <EOL> self . client . copy ( remote_path_from = self . urn . path ( ) , remote_path_to = remote_path ) <EOL> return Resource ( self . client , urn ) <EOL> def info ( self , params = None ) : <EOL> info = self . client . info ( self . urn . path ( ) ) <EOL> if not params : <EOL> return info <EOL> return { key : value for ( key , value ) in info . items ( ) if key in params } <EOL> def clean ( self ) : <EOL> return self . client . clean ( self . urn . path ( ) ) <EOL> def check ( self ) : <EOL> return self . client . check ( self . urn . path ( ) ) <EOL> def read_from ( self , buff ) : <EOL> self . client . upload_from ( buff = buff , remote_path = self . urn . path ( ) ) <EOL> def read ( self , local_path ) : <EOL> return self . client . upload_sync ( local_path = local_path , remote_path = self . urn . path ( ) ) <EOL> def read_async ( self , local_path , callback = None ) : <EOL> return self . client . upload_async ( local_path = local_path , remote_path = self . urn . path ( ) , callback = callback ) <EOL> def write_to ( self , buff ) : <EOL> return self . client . download_to ( buff = buff , remote_path = self . urn . path ( ) ) <EOL> def write ( self , local_path ) : <EOL> return self . client . download_sync ( local_path = local_path , remote_path = self . urn . path ( ) ) <EOL> def write_async ( self , local_path , callback = None ) : <EOL> return self . client . download_async ( local_path = local_path , remote_path = self . urn . path ( ) , callback = callback ) <EOL> def publish ( self ) : <EOL> return self . client . publish ( self . urn . path ( ) ) <EOL> def unpublish ( self ) : <EOL> return self . client . unpublish ( self . urn . path ( ) ) <EOL> @ property <EOL> def property ( self , option ) : <EOL> return self . client . get_property ( remote_path = self . urn . path ( ) , option = option ) <EOL> @ property . setter <EOL> def property ( self , option , value ) : <EOL> option [ '<STR_LIT:value>' ] = value . __str__ ( ) <EOL> self . client . set_property ( remote_path = self . urn . path ( ) , option = option ) </s>
