415
412
raise NotImplementedError(self.external_url)
414
def get_segment_parameters(self):
415
"""Return the segment parameters for the top segment of the URL.
417
return self._segment_parameters
419
def set_segment_parameter(self, name, value):
420
"""Set a segment parameter.
422
:param name: Segment parameter name (urlencoded string)
423
:param value: Segment parameter value (urlencoded string)
427
del self._segment_parameters[name]
431
self._segment_parameters[name] = value
432
self.base = urlutils.join_segment_parameters(
433
self._raw_base, self._segment_parameters)
417
435
def _pump(self, from_file, to_file):
418
436
"""Most children will need to copy from one file-like
419
437
object or string to another one.
485
503
raise NotImplementedError(self.abspath)
487
def _combine_paths(self, base_path, relpath):
488
"""Transform a Transport-relative path to a remote absolute path.
490
This does not handle substitution of ~ but does handle '..' and '.'
495
t._combine_paths('/home/sarah', 'project/foo')
496
=> '/home/sarah/project/foo'
497
t._combine_paths('/home/sarah', '../../etc')
499
t._combine_paths('/home/sarah', '/etc')
502
:param base_path: urlencoded path for the transport root; typically a
503
URL but need not contain scheme/host/etc.
504
:param relpath: relative url string for relative part of remote path.
505
:return: urlencoded string for final path.
507
if not isinstance(relpath, str):
508
raise errors.InvalidURL(relpath)
509
if relpath.startswith('/'):
512
base_parts = base_path.split('/')
513
if len(base_parts) > 0 and base_parts[-1] == '':
514
base_parts = base_parts[:-1]
515
for p in relpath.split('/'):
517
if len(base_parts) == 0:
518
# In most filesystems, a request for the parent
519
# of root, just returns root.
526
path = '/'.join(base_parts)
527
if not path.startswith('/'):
531
505
def recommended_page_size(self):
532
506
"""Return the recommended page size for this transport.
1369
1343
if not base.endswith('/'):
1372
self._user, self._password,
1373
self._host, self._port,
1374
self._path) = self._split_url(base)
1345
self._parsed_url = self._split_url(base)
1375
1346
if _from_transport is not None:
1376
1347
# Copy the password as it does not appear in base and will be lost
1377
1348
# otherwise. It can appear in the _split_url above if the user
1378
1349
# provided it on the command line. Otherwise, daughter classes will
1379
1350
# prompt the user for one when appropriate.
1380
self._password = _from_transport._password
1351
self._parsed_url.password = _from_transport._parsed_url.password
1352
self._parsed_url.quoted_password = (
1353
_from_transport._parsed_url.quoted_password)
1382
base = self._unsplit_url(self._scheme,
1383
self._user, self._password,
1384
self._host, self._port,
1355
base = str(self._parsed_url)
1387
1357
super(ConnectedTransport, self).__init__(base)
1388
1358
if _from_transport is None:
1441
1429
def relpath(self, abspath):
1442
1430
"""Return the local path portion from a given absolute path"""
1443
scheme, user, password, host, port, path = self._split_url(abspath)
1431
parsed_url = self._split_url(abspath)
1445
if (scheme != self._scheme):
1433
if parsed_url.scheme != self._parsed_url.scheme:
1446
1434
error.append('scheme mismatch')
1447
if (user != self._user):
1435
if parsed_url.user != self._parsed_url.user:
1448
1436
error.append('user name mismatch')
1449
if (host != self._host):
1437
if parsed_url.host != self._parsed_url.host:
1450
1438
error.append('host mismatch')
1451
if (port != self._port):
1439
if parsed_url.port != self._parsed_url.port:
1452
1440
error.append('port mismatch')
1453
if not (path == self._path[:-1] or path.startswith(self._path)):
1441
if (not (parsed_url.path == self._parsed_url.path[:-1] or
1442
parsed_url.path.startswith(self._parsed_url.path))):
1454
1443
error.append('path mismatch')
1456
1445
extra = ', '.join(error)
1457
1446
raise errors.PathNotChild(abspath, self.base, extra=extra)
1458
pl = len(self._path)
1459
return path[pl:].strip('/')
1447
pl = len(self._parsed_url.path)
1448
return parsed_url.path[pl:].strip('/')
1461
1450
def abspath(self, relpath):
1462
1451
"""Return the full url to the given relative path.
1584
1567
raise NotImplementedError(self.disconnect)
1587
def get_transport(base, possible_transports=None):
1588
"""Open a transport to access a URL or directory.
1590
:param base: either a URL or a directory name.
1592
:param transports: optional reusable transports list. If not None, created
1593
transports will be added to the list.
1595
:return: A new transport optionally sharing its connection with one of
1596
possible_transports.
1570
def location_to_url(location):
1571
"""Determine a fully qualified URL from a location string.
1573
This will try to interpret location as both a URL and a directory path. It
1574
will also lookup the location in directories.
1576
:param location: Unicode or byte string object with a location
1577
:raise InvalidURL: If the location is already a URL, but not valid.
1578
:return: Byte string with resulting URL
1580
if not isinstance(location, basestring):
1581
raise AssertionError("location not a byte or unicode string")
1601
1582
from bzrlib.directory_service import directories
1602
base = directories.dereference(base)
1604
def convert_path_to_url(base, error_str):
1605
if urlutils.is_url(base):
1606
# This looks like a URL, but we weren't able to
1607
# instantiate it as such raise an appropriate error
1608
# FIXME: we have a 'error_str' unused and we use last_err below
1609
raise errors.UnsupportedProtocol(base, last_err)
1610
# This doesn't look like a protocol, consider it a local path
1611
new_base = urlutils.local_path_to_url(base)
1612
# mutter('converting os path %r => url %s', base, new_base)
1583
location = directories.dereference(location)
1615
1585
# Catch any URLs which are passing Unicode rather than ASCII
1617
base = base.encode('ascii')
1587
location = location.encode('ascii')
1618
1588
except UnicodeError:
1619
# Only local paths can be Unicode
1620
base = convert_path_to_url(base,
1621
'URLs must be properly escaped (protocol: %s)')
1589
if urlutils.is_url(location):
1590
raise errors.InvalidURL(path=location,
1591
extra='URLs must be properly escaped')
1592
location = urlutils.local_path_to_url(location)
1594
if location.startswith("file:") and not location.startswith("file://"):
1595
return urlutils.join(urlutils.local_path_to_url("."), location[5:])
1597
if not urlutils.is_url(location):
1598
return urlutils.local_path_to_url(location)
1603
def get_transport_from_path(path, possible_transports=None):
1604
"""Open a transport for a local path.
1606
:param path: Local path as byte or unicode string
1607
:return: Transport object for path
1609
return get_transport_from_url(urlutils.local_path_to_url(path),
1610
possible_transports)
1613
def get_transport_from_url(url, possible_transports=None):
1614
"""Open a transport to access a URL.
1617
:param transports: optional reusable transports list. If not None, created
1618
transports will be added to the list.
1620
:return: A new transport optionally sharing its connection with one of
1621
possible_transports.
1623
1623
transport = None
1624
1624
if possible_transports is not None:
1625
1625
for t in possible_transports:
1626
t_same_connection = t._reuse_for(base)
1626
t_same_connection = t._reuse_for(url)
1627
1627
if t_same_connection is not None:
1628
1628
# Add only new transports
1629
1629
if t_same_connection not in possible_transports:
1630
1630
possible_transports.append(t_same_connection)
1631
1631
return t_same_connection
1633
1634
for proto, factory_list in transport_list_registry.items():
1634
if proto is not None and base.startswith(proto):
1635
transport, last_err = _try_transport_factories(base, factory_list)
1635
if proto is not None and url.startswith(proto):
1636
transport, last_err = _try_transport_factories(url, factory_list)
1637
1638
if possible_transports is not None:
1638
1639
if transport in possible_transports:
1639
1640
raise AssertionError()
1640
1641
possible_transports.append(transport)
1641
1642
return transport
1643
# We tried all the different protocols, now try one last time
1644
# as a local protocol
1645
base = convert_path_to_url(base, 'Unsupported protocol: %s')
1647
# The default handler is the filesystem handler, stored as protocol None
1648
factory_list = transport_list_registry.get(None)
1649
transport, last_err = _try_transport_factories(base, factory_list)
1643
if not urlutils.is_url(url):
1644
raise errors.InvalidURL(path=url)
1645
raise errors.UnsupportedProtocol(url, last_err)
1648
def get_transport(base, possible_transports=None):
1649
"""Open a transport to access a URL or directory.
1651
:param base: either a URL or a directory name.
1653
:param transports: optional reusable transports list. If not None, created
1654
transports will be added to the list.
1656
:return: A new transport optionally sharing its connection with one of
1657
possible_transports.
1661
return get_transport_from_url(location_to_url(base), possible_transports)
1654
1664
def _try_transport_factories(base, factory_list):
added
removed
bzrlib/transport/__init__.py
