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

Subclasses

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

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

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

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