Module `pyfx.internal.connection`

Interacting with a Pricefx platform directly.

This platform defines a low-level wrapper to interact with the Pricefx platform. For a higher level API, see the pyfx.api.domain package.

Classes

class Connection

Abstract class for a Connection.

A connection abstract the low-lever details of interacting with a "backend". This backend can be an actual Pricefx instance, or can be substituted with, e.g. a local mock.

Ancestors

abc.ABC

Methods

def add_object(self, type_code: str, attributes: Dict[str, Any]) ‑> Dict[str, Any]

Add an object and returns its actual attributes.

def append_to_table(self, typedid: str, data: AvroStream) ‑> None

Append data to an already existing DMTable or DMDataSource.

Args

typedid: the typed id of the table to append
data: the content of the data to push in avro format

def attach_file(self, typedid: str, name: str, content: ) ‑> None

Attach a content as a file with a given name to the model.

def create_table(self, name: str, fields_spec: List[Dict], content: AvroStream, label: Optional[str] = None, owner_typedid: Optional[str] = None, replace_existing: bool = True) ‑> None

Create a table in the backend.

Args

name: the name of the table to push
fields_spec: a list of dictionaries that describe the columns of the data source

[
   { name:"productId", label:"Product ID", type:"TEXT", key:true},
   { name:"productGroup", label:"ProductGroup", type:"TEXT", dimension:true},
   { name:"revenue", label:"Revenue", type:"MONEY"},
]

content: the content of the source file to push in avro format
label: the label of the table (optional, defaults to name)
owner_typedid: the name of the table owner (optional, set it to create a DMT)
replace_existing: if True then removes the preceding data source having the same name if it exists before pushing the data (default = True).

def get_calcitems(self, typedid: str) ‑> List[Dict[str, Any]]

Get the calculation items associated with a specific typedid.

def get_fcs(self, typedid: str, params: Optional[Dict[str, Any]] = None) ‑> Dict[str, Any]

Get the attributes of a specific fields collection.

def get_object(self, typedid: str) ‑> Dict[str, Any]

Get the attributes of a specific element.

def list_attachments(self, typedid: str) ‑> List[Dict[str, Any]]

Get the information about attachments of the model.

def list_fcs(self, type_code: str, params: Optional[Dict[str, Any]] = None) ‑> List[Dict[str, Any]]

List all the fields collections of a given type.

def list_objects(self, type_code: str) ‑> List[Dict[str, Any]]

List all the elements of a given type.

def pull_file(self, owner_typedid: str, attachment_typedid: str, chunk_size: int = 128) ‑> Iterator[bytes]

Fetch file from the backend.

def push_calcitem(self, typedid: str, key1: str, key2: str, value: Any) ‑> None

Push a new calculation item associated with a specific typedid.

def stream_fcs(self, typedid: str, chunk_size: int = 128) ‑> Iterator[bytes]

Stream the content of a data source.

def update_object(self, type_code: str, attributes: Dict[str, Any]) ‑> Dict[str, Any]

Update an object and returns its actual attributes.

def update_status(self, jst_id: int, status_code: JobStatus, progress: Optional[int], msg: Optional[str] = None, results: Optional[Dict[str, Any]] = None) ‑> None

Update the job status on the backend.

Args

jst_id: the id of the jst
status_code: the new job status
progress: the new job progress percent (optional)
msg: a message for the new status change (optional)
results: a dictionary associating result names to their value (optional). Will be available to next evaluations and calculations.

class ConnectionComposed (dispatch: Dict[str, Connection], default_connection: Connection)

A connection that composes a remote and local connections.

This composite connection allows to mix and match calls between a remote and local connection according to a given configuration.

Ancestors

Connection
abc.ABC

Methods

def add_object(self, type_code: str, attributes: Dict[str, Any]) ‑> Dict[str, Any]: See Connection corresponding method.
def append_to_table(self, typedid: str, data: AvroStream) ‑> None: See Connection corresponding method.
def attach_file(self, typedid: str, name: str, content: ) ‑> None: See Connection corresponding method.
def create_table(self, name: str, fields_spec: List[Dict], content: AvroStream, label: Optional[str] = None, owner_typedid: Optional[str] = None, replace_existing: bool = True) ‑> None: See Connection corresponding method.
def get_calcitems(self, typedid: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def get_fcs(self, typedid: str, params: Optional[Dict[str, Any]] = None) ‑> Dict[str, Any]: See Connection corresponding method.
def get_object(self, typedid: str) ‑> Dict[str, Any]: See Connection corresponding method.
def list_attachments(self, typedid: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def list_fcs(self, type_code: str, params: Optional[Dict[str, Any]] = None) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def list_objects(self, type_code: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def pull_file(self, owner_typedid: str, attachment_typedid: str, chunk_size: int = 128) ‑> Iterator[bytes]: See Connection corresponding method.
def push_calcitem(self, typedid: str, key1: str, key2: str, value: Any) ‑> None: See Connection corresponding method.
def stream_fcs(self, typedid: str, chunk_size: int = 128) ‑> Iterator[bytes]: See Connection corresponding method.
def update_object(self, type_code: str, attributes: Dict[str, Any]) ‑> Dict[str, Any]: See Connection corresponding method.
def update_status(self, jst_id: int, status_code: JobStatus, progress: Optional[int], msg: Optional[str] = None, results: Optional[Dict[str, Any]] = None) ‑> None: See Connection corresponding method.

class ConnectionLocal (path: pathlib.Path, logformat: Optional[str] = None)

A connection that use the local filesystem.

This "connection" actually pull/push data from/to the local filesystem. Useful debugging or runing scripts locally.

Ancestors

Connection
abc.ABC

Methods

def add_object(self, type_code: str, attributes: Dict[str, Any]) ‑> Dict[str, Any]: See Connection corresponding method.

This implementation of the methods always returns the attributes.
def append_to_table(self, typedid: str, data: AvroStream) ‑> None: See Connection corresponding method.
def attach_file(self, typedid: str, name: str, content: ) ‑> None: See Connection corresponding method.
def create_table(self, name: str, fields_spec: List[Dict], content: AvroStream, label: Optional[str] = None, owner_typedid: Optional[str] = None, replace_existing: bool = True) ‑> None: See Connection corresponding method.
def get_calcitems(self, typedid: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def get_fcs(self, typedid: str, params: Optional[Dict[str, Any]] = None) ‑> Dict[str, Any]: See Connection corresponding method.

This implementation will returns a minimal dict.
def get_object(self, typedid: str) ‑> Dict[str, Any]: See Connection corresponding method.

This implementation will returns a minimal dict.
def list_attachments(self, typedid: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def list_fcs(self, type_code: str, params: Optional[Dict[str, Any]] = None) ‑> List[Dict[str, Any]]: See Connection corresponding method.

This implementation of the methods always returns an empty list.
def list_objects(self, type_code: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.

This implementation of the methods always returns an empty list.
def pull_file(self, owner_typedid: str, attachment_typedid: str, chunk_size: int = 128) ‑> Iterator[bytes]: See Connection corresponding method.
def push_calcitem(self, typedid: str, key1: str, key2: str, value: Any) ‑> None: See Connection corresponding method.
def stream_fcs(self, typedid: str, chunk_size: int = 128) ‑> Iterator[bytes]: See Connection corresponding method.
def update_object(self, type_code: str, attributes: Dict[str, Any]) ‑> Dict[str, Any]: See Connection corresponding method.

This implementation of the methods always returns the attributes.
def update_status(self, jst_id: int, status_code: JobStatus, progress: Optional[int], msg: Optional[str] = None, results: Optional[Dict[str, Any]] = None) ‑> None: See Connection corresponding method.

class ConnectionRemote (endpoint: str, pfxsession: PfxSession)

A connection to a remote instance.

This class abstracts the various aspects of interacting with the Pricefx platform behind an unified interface.

Ancestors

Connection
abc.ABC

Methods

def add_object(self, type_code: str, attributes: Dict[str, Any]) ‑> Dict[str, Any]: See Connection corresponding method.
def append_to_table(self, typedid: str, data: AvroStream) ‑> None: See Connection corresponding method.
def attach_file(self, typedid: str, name: str, content: ) ‑> None: See Connection corresponding method.
def create_table(self, name: str, fields_spec: List[Dict], content: AvroStream, label: Optional[str] = None, owner_typedid: Optional[str] = None, replace_existing: bool = True) ‑> None: See Connection corresponding method.
def get_calcitems(self, typedid: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def get_fcs(self, typedid: str, params: Optional[Dict[str, Any]] = None) ‑> Dict[str, Any]: See Connection corresponding method.
def get_object(self, typedid: str) ‑> Dict[str, Any]: See Connection corresponding method.
def list_attachments(self, typedid: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def list_fcs(self, type_code: str, params: Optional[Dict[str, Any]] = None) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def list_objects(self, type_code: str) ‑> List[Dict[str, Any]]: See Connection corresponding method.
def pull_file(self, owner_typedid: str, attachment_typedid: str, chunk_size: int = 128) ‑> Iterator[bytes]: See Connection corresponding method.
def push_calcitem(self, typedid: str, key1: str, key2: str, value: Any) ‑> None: See Connection corresponding method.
def stream_fcs(self, typedid: str, chunk_size: int = 128) ‑> Iterator[bytes]: See Connection corresponding method.
def update_object(self, type_code: str, attributes: Dict[str, Any]) ‑> Dict[str, Any]: See Connection corresponding method.
def update_status(self, jst_id: int, status_code: JobStatus, progress: Optional[int], msg: Optional[str] = None, results: Optional[Dict[str, Any]] = None) ‑> None: See Connection corresponding method.

class JobStatus (value, names=None, *, module=None, qualname=None, type=None, start=1)

Possible status of a job for the Pricefx REST API.

Ancestors

enum.Enum

Class variables

var FAILED
var FINISHED
var PROCESSING