xcp.accessor

accessor - provide common interface to access methods

class xcp.accessor.Accessor(ro)[source]

Bases: object

access(name)[source]

Return boolean determining where ‘name’ is an accessible object in the target.

canEject()[source]
finish()[source]
openAddress(address: str) IO[bytes] | Literal[False][source]

must be overloaded

openText(address: str) Generator[IO[str] | Literal[False], None, None][source]

Context manager to read text from address using ‘with’. Yields IO[str] or False

start()[source]
xcp.accessor.AnyAccessor

Type alias for static typing the Accessor object returned by createAccessor()

alias of HTTPAccessor | FTPAccessor | FileAccessor | DeviceAccessor | NFSAccessor

class xcp.accessor.DeviceAccessor(device, ro, fs=None)[source]

Bases: MountingAccessor

class xcp.accessor.FTPAccessor(baseAddress, ro)[source]

Bases: Accessor

access(path)[source]

Return boolean determining where ‘name’ is an accessible object in the target.

finish()[source]
openAddress(address)[source]

must be overloaded

start()[source]
writeFile(in_fh, out_name)[source]
class xcp.accessor.FileAccessor(baseAddress, ro)[source]

Bases: Accessor

openAddress(address)[source]

must be overloaded

writeFile(in_fh, out_name)[source]
class xcp.accessor.FilesystemAccessor(location, ro)[source]

Bases: Accessor

openAddress(address)[source]

must be overloaded

class xcp.accessor.HTTPAccessor(baseAddress, ro)[source]

Bases: Accessor

openAddress(address)[source]

must be overloaded

xcp.accessor.Mount

Type alias for static typing or unit tests testing subclasses of MountingAccessor

alias of DeviceAccessor | NFSAccessor

class xcp.accessor.MountingAccessor(mount_types, mount_source, mount_options=None)[source]

Bases: FilesystemAccessor

finish()[source]
start()[source]
writeFile(in_fh, out_name)[source]
xcp.accessor.MountingAccessorTypes = (<class 'xcp.accessor.DeviceAccessor'>, <class 'xcp.accessor.NFSAccessor'>)

Tuple for type checking in unit tests testing subclasses of MountingAccessor

class xcp.accessor.NFSAccessor(nfspath, ro)[source]

Bases: MountingAccessor

xcp.accessor.SUPPORTED_ACCESSORS: dict[str, type[HTTPAccessor | FTPAccessor | FileAccessor | DeviceAccessor | NFSAccessor]] = {'dev': <class 'xcp.accessor.DeviceAccessor'>, 'file': <class 'xcp.accessor.FileAccessor'>, 'ftp': <class 'xcp.accessor.FTPAccessor'>, 'http': <class 'xcp.accessor.HTTPAccessor'>, 'https': <class 'xcp.accessor.HTTPAccessor'>, 'nfs': <class 'xcp.accessor.NFSAccessor'>}

Dict of supported accessors. The key is the URL scheme

xcp.accessor.createAccessor(baseAddress: str, *args: bool | Tuple[bool, List[str]]) Literal[False] | AnyAccessor[source]

Return instance of the appropriate Accessor subclass based on the baseAddress.

Parameters:
  • (str) (baseAddress) – The base address for the accessor.

  • (tuple) (args) – Additional argument(s) to be passed to the accessor constructor

Returns Accessor (object | Literal[False]):

Accessor object or Literal[False]

Raises:

AssertionError – If the scheme of the baseAddress is not supported.

Also raises AssertionError when baseAddress is file:///filename/ but the final / is omitted. The final terminating / is compulsory.

For all Accessors, the 1st arg after the address is type bool for ro (readonly flag) The DeviceAccessor accepts a 3rd argument: a List[] of filesystem names

Example usage:

accessor = createAccessor("http://example.com", True)
accessor = createAccessor("dev://example.com", True, ['iso9660', 'ext3'])
if not accessor:
    fatal()
else:
    accessor.read()
xcp.accessor.mapError(errorCode)[source]
xcp.accessor.rebuild_url(url_parts)[source]

Rebuild URL without auth components