ocx-common

clients

Module for server clients.

exception ocx_common.clients.clients.CurlClientError

Bases: error

Curl client errors.

class ocx_common.clients.clients.CurlRestClient(headers=None, timeout=30)

Bases: IRestClient, ABC

cURL client

Parameters:

• headers – The request headers

• timeout – The request timeout

Parameters:

• base_url – base url

• timeout – Timeout in ms

delete(url, payload, headers=None)

Delete method.

Parameters:

• headers – The request header

• url – the resource.

• payload – The request body.

Return type:

Response

Returns:

The Response object.

Raises:

RequestClientError on (HTTPError, ConnectionError). –

get(url, headers=None)

Get method.

Parameters:

• headers – Request header

• url – The resource endpoint

Return type:

Response

Returns:

The Response object.

Raises:

CurlClientError if the status code is not 200. –

patch(url, payload, headers=None)

Patch method.

Parameters:

• headers – The request header

• url – the resource.

• payload – The request body.

Return type:

Response

Returns:

The Response object.

Raises:

RequestClientError on (HTTPError, ConnectionError). –

post(url, payload, headers=None)

Post method.

Parameters:

• headers – The request headers

• url – the resource.

• payload – The request body.

Return type:

Response

Returns:

The Response object.

Raises:

RequestClientError on pycurl.error. –

put(url, payload, headers=None)

Put method.

Parameters:

• headers – Request header

• url – the resource.

• payload – The request body.

Return type:

Response

Returns:

The Response object.

Raises:

RequestClientError on (HTTPError, ConnectionError). –

pycurl_to_requests_response(buffer, status_code)

A function to convert pycurl response to a requests Response object

Return type:

Response

set_headers(headers)

Set the request headers :Parameters: headers – The headers declaration

class ocx_common.clients.clients.EmbeddingMethod(value)

Bases: Enum

Embedding type.

BASE64 = 'BASE64'

STRING = 'STRING'

URL = 'URL'

exception ocx_common.clients.clients.RequestClientError(*args, **kwargs)

Bases: RequestException

Request client errors.

Initialize RequestException with request and response objects.

class ocx_common.clients.clients.RestClient(headers=None, timeout=30)

Bases: IRestClient, ABC

Request client

Parameters:

• headers – The request headers

• timeout – The request timeout

Parameters:

• base_url – base url

• timeout – Timeout in ms

delete(url, payload, headers=None)

Delete method.

Parameters:

• headers – The request header

• url – the resource.

• payload – The request body.

Return type:

Response

Returns:

The Response object.

Raises:

RequestClientError on (HTTPError, ConnectionError). –

get(url, headers=None)

Get method.

Parameters:

• headers – The request header dictionary

• url – The resource url

Return type:

Response

Returns:

The request Response object.

Raises:

RequestClientError on (HTTPError, ConnectionError). –

patch(url, payload, headers=None)

Patch method.

Parameters:

• headers – The request header

• url – the resource.

• payload – The request body.

Return type:

Response

Returns:

The Response object.

Raises:

RequestClientError on (HTTPError, ConnectionError). –

post(url, payload, headers=None)

Post method.

Parameters:

• headers – The request headers

• url – the resource.

• payload – The request body.

Return type:

Response

Returns:

The Response object.

Raises:

RequestClientError on (HTTPError, ConnectionError). –

put(url, payload, headers=None)

Put method.

Parameters:

• headers – Request header

• url – the resource.

• payload – The request body.

Return type:

Response

Returns:

The Response object.

Raises:

RequestClientError on (HTTPError, ConnectionError). –

set_headers(headers)

Set the request headers

Parameters:

headers – The headers declaration

decorators

Reusable decorators

ocx_common.decorators.decorators.debugger(func)

@debug decorator using ice cream

ocx_common.decorators.decorators.exception_handler(exception_to_raise)

@exception_handler decorator

ocx_common.decorators.decorators.memoize(func)

@memoize decorator

ocx_common.decorators.decorators.timer(func)

@timer decorator for measuring elapsed time of a function call.

interfaces

Interfaces module.

class ocx_common.interfaces.interfaces.IModuleDeclaration

Bases: ABC

Abstract module import declaration Interface

abstract get_declaration()

Abstract Method: Return the module declaration string.

Return type:

str

class ocx_common.interfaces.interfaces.IObservable

Bases: ABC

Interface. The observable object.

abstract subscribe(observer)

subscription

abstract unsubscribe(observer)

abstract update(event, message)

update method.

Parameters:

• event – The event type

• message – The event message

class ocx_common.interfaces.interfaces.IObserver

Bases: ABC

The observer interface

abstract update(event, payload)

Interface update method

class ocx_common.interfaces.interfaces.IParser

Bases: ABC

Abstract IParser interface.

abstract iterator(model)

Abstract method for iterating a data model.

Parameters:

model – the data model to iterate on.

Return type:

Iterator

Returns:

An iterator

abstract parse(model)

Abstract method for parsing a data model,

Parameters:

model – the data model source

Return type:

Optional[Any]

Returns:

the root dataclass of the parsed data model.

class ocx_common.interfaces.interfaces.IRestClient(headers, timeout=30)

Bases: ABC

Abstract IRestClient interface for REST server clients.

Parameters:

• headers – The request headers

• timeout – The request timeout

Parameters:

• base_url – base url

• timeout – Timeout in ms

abstract delete(url, payload, headers)

Abstract delete method

Return type:

Response

abstract get(url, headers)

Abstract get method

Return type:

Response

abstract patch(url, payload, headers)

Abstract patch method

Return type:

Response

abstract post(url, payload, headers)

Abstract post method

Return type:

Response

abstract put(url, payload, headers)

Abstract put method

Return type:

Response

abstract set_headers(headers)

Abstract set headers method

class ocx_common.interfaces.interfaces.ISerializer(model)

Bases: ABC

OcxSerializer interface

abstract serialize_to_file(to_file)

Abstract XML serialize to file method

Return type:

bool

abstract serialize_to_string()

Abstract XML serialize to string method

Return type:

str

loader

Dynamically load a python module.

class ocx_common.loader.loader.DeclarationOfOcxImport(name, version)

Bases: IModuleDeclaration

Declaration of the ocx module.

get_declaration()

Return the module import declaration.

Return type:

str

get_name()

Return the declared module name.

Return type:

str

get_version()

Return the OCX module version.

Return type:

str

class ocx_common.loader.loader.DynamicLoader

Bases: object

Dynamically loads modules, classes of functions from a module declaration.

classmethod get_all_class_names(module_name, version)

Return all class names in the module by the __all__ variable.

Parameters:

• module_name – The module name

• version – The module version

Return type:

List

Returns:

The list of available module class names.

Example

>>> from ocx_common.loader.loader import DeclarationOfOcxImport , DynamicLoader
>>> class_name = "Vessel"
>>> declaration = DeclarationOfOcxImport("ocx", '3.0.1')
>>> data_class = DynamicLoader.import_class(declaration, class_name)()
>>> data_class.__doc__
Vessel asset subject to Classification.

classmethod import_class(module_declaration, class_name)

The module import declaration.

Parameters:

• class_name – The class name to load form the declared module

• module_declaration – The declaration of the python module to be loaded

Return type:

Any

Returns:

Return the loaded class, None if failed.

classmethod import_module(module_declaration)

Parameters:

module_declaration – The declaration of the python module to load

Return type:

ModuleType

Returns:

Return the loaded module, None if failed.

exception ocx_common.loader.loader.DynamicLoaderError

Bases: BaseException

Dynamic import errors.

class ocx_common.loader.loader.ModuleDeclaration(package, sub_module, name)

Bases: IModuleDeclaration, ABC

General module declaration

Parameters:

• package – the package name

• sub_module – The submodule-name

• name – The method name

get_declaration()

Return the module import declaration.

Return type:

str

parser

Module for parsing a 3Docx model.

class ocx_common.parser.parsers.MetaData

Bases: object

Dataclass metadata.

static class_name(data_class)

Return the name of the class

Return type:

str

static meta_class_fields(data_class)

Return the dataclass metadata.

Parameters:

data_class – The dataclass instance

Return type:

Dict

Returns:

The metadata of the class

static name(data_class)

Get the OCX name.

Parameters:

data_class – The dataclass instance

Return type:

str

Returns:

The name of the OCX type

static namespace(data_class)

Get the OCX namespace

Parameters:

data_class – The dataclass instance

Return type:

str

Returns:

The namespace of the dataclass

class ocx_common.parser.parsers.ObservableEvent(value)

Bases: Enum

Events that can be listened to and broadcast.

DATACLASS = 'dataclass'

REPORT = 'report'

SERIALIZE = 'serialize'

class ocx_common.parser.parsers.OcxModelParser(fail_on_unknown_properties=False, fail_on_unknown_attributes=False, fail_on_converter_warnings=True)

Bases: object

OcxModelParser class for 3Docx XML files.

Parameters:

ocx_model – The file path or URL to the source model.

get_root()

Return the XML document root.

Return type:

Optional[Element]

parse(xml_file)

Parse the XML file and return the root element. Arg:

• model: The location of the XML file

Return type:

Optional[Element]

Returns:

The etree root

Raises:

OcxParserError –

parse_element(element, clazz)

Parse a 3Docx XML element and return the dataclass.

Parameters:

• - element – The 3Docx XML Element to parse.

• - clazz – The class to be instantiated

Return type:

Optional[TypeVar(T)]

Returns:

The element dataclass instance if created, else returns None.

class ocx_common.parser.parsers.OcxNotifyParser(fail_on_unknown_properties=False, fail_on_unknown_attributes=False, fail_on_converter_warnings=True)

Bases: IObservable, ABC

Ocx notification parser class for 3Docx XML files.

Parameters:

• fail_on_unknown_properties – Don’t bail out on unknown properties.

• fail_on_unknown_attributes – Don’t bail out on unknown attributes

• fail_on_converter_warnings – bool = Convert warnings to exceptions

class_factory(clazz, params)

Custom class factory method

parse(xml_file)

Parse a 3Docx XML model and return the root dataclass.

Parameters:

xml_file – The 3Docx xml file or url to parse.

Return type:

Optional[TypeVar(T)]

Returns:

The root dataclass instance of the parsed 3Docx XML.

parse_element(element, ocx_module)

Parse a 3Docx XML element and return the dataclass.

Parameters:

element – The 3Docx XML Element to parse.

Return type:

Optional[TypeVar(T)]

Returns:

The element dataclass instance if created, else returns None.

subscribe(observer)

subscription

unsubscribe(observer)

update(event, payload)

update method.

Parameters:

• event – The event type

• message – The event message

exception ocx_common.parser.parsers.OcxParserError

Bases: ParserError

Errors raised by this module.

exception ocx_common.parser.parsers.XmlParserError

Bases: ValueError

Parser errors.

class ocx_common.parser.parsers.XsdSchemaParser(target_namespace, location)

Bases: object

A schema parser parsing an XSD schema and returning the xsdata.models.xsd.Schema.

Parameters:

• - target_namespace – The target namespace of the schema.

• - location – The path to the schema XSD file.

parse_from_bytes(input_stream)

Parse the XSD from bytes and return the Schema object.

Parameters:

input_stream

Return type:

Schema

parse_from_path(path)

Parse the XSD and return the Schema object

Return type:

Schema

parse_from_string(input_string)

Parse the XSD and return the Schema object

Return type:

Schema

utilities

Shared utility classes and functions

ocx_common.utilities.utilities.all_equal(iterable)

Verify that all items in a list are equal :Parameters: iterable

Return type:

bool

Returns:

True if all are equal, False otherwise.

ocx_common.utilities.utilities.camel_case_split(str)

Split camel case string to individual strings.

Return type:

List

ocx_common.utilities.utilities.default_to_grid(d)

Converts defaultdicts to a data grid with unique row ids.

Parameters:

d – The dict to be converted

Return type:

Dict

ocx_common.utilities.utilities.default_to_regular(d)

Converts defaultdict of defaultdict to dict of dicts.

Parameters:

d – The dict to be converted

Return type:

Dict

ocx_common.utilities.utilities.dromedary_case_split(str)

Split camel case string to individual strings.

Return type:

List

ocx_common.utilities.utilities.file_uri_to_path(uri)

Converts a file:// URI to a proper file system path.

Return type:

str

ocx_common.utilities.utilities.get_file_path(file_name)

Get the correct file path also when called within a one-file executable.

ocx_common.utilities.utilities.get_key_from_value(my_dict, value)

Return the key associated with the value

Parameters:

• my_dict – The dictionary of key, value pairs

• value – The value to search for

Return type:

Any

Returns:

The key if found

Raises:

ValueError if the value does not exist in the dictionary. –

ocx_common.utilities.utilities.is_local_file_uri(uri)

Return True if the file uri is a local file

Return type:

bool

ocx_common.utilities.utilities.is_substring_in_list(substring, string_list)

Parameters:

• substring – The search string

• string_list – List of strings

Returns:

True if the substring is found, False otherwise.

ocx_common.utilities.utilities.is_valid_absolute_windows_path(path)

Returns True if the path is a valid absolute Windows path.

Parameters:

path (str) – The path to validate.

Return type:

bool

Returns:

bool – True if valid, False otherwise.

ocx_common.utilities.utilities.is_valid_file_path(path)

Parameters:

path

Returns:

Return type:

bool

ocx_common.utilities.utilities.is_valid_unix_file_path(path)

Return type:

bool

Return True if the path is a valid UNIX path

Parameters:

path – path to validate

Returns: True if validated, False otherwise.

ocx_common.utilities.utilities.is_windows_drive_letter(scheme)

Parameters:

scheme – The urlparse scheme

Return type:

bool

Returns:

True if the scheme is a valid Windows drive letter

ocx_common.utilities.utilities.iter_files(directory, filter_str)

Iterate over files in a directory based on a specified filter pattern.

Parameters:

• directory (str) – The path to the directory.

• filter_str (str) – The filter pattern to apply when filtering files.

Return type:

Generator

Returns:

Generator – A generator yielding file paths that match the filter criteria.

ocx_common.utilities.utilities.list_files_in_directory(directory, filter)

Lists files in a directory based on a specified filter using glob pattern matching.

Parameters:

• directory (str) – The path to the directory.

• filter (str) – The filter pattern to apply when listing files.

Return type:

List

Returns:

List – A list of file paths that match the filter criteria.

Raises:

AssertionError – If the directory does not exist.

ocx_common.utilities.utilities.nested_dict()

A recursive function that creates a default dictionary where each value is another default dictionary.

ocx_common.utilities.utilities.parent_directory(file)

The full path to the folder containing the file

Parameters:

file – The name of an existing file

Return type:

str

ocx_common.utilities.utilities.resource_path(relative_path)

Get absolute path to resource, works for dev and for PyInstaller

x_path

A python XPath implementation for OCX types

class ocx_common.x_path.x_path.OcxPath(document_root, namespaces, extensions=None, regexp=True, smart_strings=True)

Bases: object

get_all_named_children(node, child_name, namespace='ocx')

Return type:

List[Any]

get_all_named_ocx_elements(name, namespace='ocx')

Return type:

List[Any]

get_ocx_attribute_value_collection(element, attribute_name, namespace='ocx')

Parameters:

• element – The lxml Element instance

• attribute_name – The attribute name to be retrieved

• namespace – Optional namespace prefix. Default = “ocx”

Return type:

List[Any]

Returns:

A list of attributes.

get_ocx_attribute_with_value(element, attribute_name, attribute_value, namespace='ocx')

Parameters:

• element

• attribute_name

• attribute_value

• namespace

Returns:

Return type:

List[Any]

class ocx_common.x_path.x_path.OcxPathBuilder

Bases: object

static select_all_named_nodes(nodename, namespace='ocx')

Return type:

str

static select_any_nodes_with_attribute_value(attribute_name, attribute_value, namespace='ocx')

Return type:

str

static select_any_nodes_with_global_attribute_name(attribute_name, namespace='ocx')

Return type:

str

static select_any_nodes_with_local_attribute_name(attribute_name, namespace='ocx')

Return type:

str

static select_any_nodes_with_local_value(attribute_name, attribute_value, namespace='ocx')

Return type:

str

static select_current_node()

Return type:

str

static select_named_nodes(node_name, namespace='ocx')

Return type:

str

static select_named_nodes_with_global_attribute_name(node_name, attribute_name, namespace='ocx')

Return type:

str

static select_named_nodes_with_global_attribute_value(node_name, attribute_name, attribute_value, namespace='ocx')

Return type:

str

static select_named_nodes_with_local_attribute_name(node_name, attribute_name, namespace='ocx')

Return type:

str

static select_named_nodes_with_local_attribute_value(node_name, attribute_name, attribute_value, namespace='ocx')

Return type:

str

static select_named_parent_node(parent, namespace='ocx')

Return type:

str

static select_parent_node()

Return type:

str

exception ocx_common.x_path.x_path.OcxPathError

Bases: ValueError, XPathError