xia_prompts_metagpt.prompts.MetagptPrompts

class xia_prompts_metagpt.prompts.MetagptPrompts(**kwargs)

Bases: Template

Mark the usage of current GPT Templates

__init__(**kwargs)

Methods

`__init__`(**kwargs)
`action`(action_name[, acl, payload])	Doing an action of a document
`analyze`(analytic_request[, acl])	Give analyzing results
`backup`([location, data_encode, data_format, ...])	Backup data of a model
`calculate_id`()	Calculate document id from current attributes
`calculate_jinja_env`(resource_type, ...)
`check_acl`(acl, act[, doc, raise_error])	Check user authorization
`check_dependency`(old_data[, new_data, batch])	Check if this document supports other document
`check_scope`(scope)	Check if data match the data scope
`check_update_acl`(_acl[, _doc])	Special ACL check for update.
`collection_action`(action_name[, acl, payload])	Doing a collection level action
`create_collection`([acl])	Create the collection
`delete`([batch, acl])	Delete a document
`delete_all`([acl])	Delete every instance of this document type
`dict_to_id`(key_values)	Calculate document id from key_values
`dict_to_id_list`(key_values)	Calculate document id from key_values.
`drop`([acl])	Drop the collection
`from_db`([_engine])	Transform database data to internal data
`from_display`(**kwargs)	Transform Display data to internal data
`get_actions`()	Get action supported by this document
`get_address`([engine])	Get address for an engine
`get_all_fields`()	Get All fields of current class
`get_collection_name`([engine])	Get collection name of current class
`get_display_data`([lazy, catalog, show_hidden])	Get Display data (visualized form of data)
`get_field`(parent_object, field_path)	Get field object from a field path
`get_id`()	Get document id
`get_jinja_env`(resource_type)
`get_meta_data`()	Get Meta-data of a table
`get_raw_data`()	Get Raw data (materialized form of data)
`get_replica_engines`()	Get engines for the keeping replicas
`get_runtime_data`()	Get Runtime data (calculated from raw data when runtime is activated)
`get_sample`()	Get a sample document data
`get_version`(doc_id)	Get a version code (uuid) for the given object
`id_to_dict`(doc_id)	From document id to key
`load`(*args[, _batch, _acl])	Load a Document from engine
`lock`([timeout, acl])	Lock the document for update
`migrate`(original_doc[, copy_data])	Migrate a document from the original one into new one.
`objects`(*args[, _batch, _acl, _limit, _id_only])	Search documents
`purge_version_table`()	Purge version table
`reload`([batch, acl])	Reload the document from the engine
`replicate`(task_list[, acl])	Replicate documents
`restore`([location, data_encode, ...])	Restore data of a model
`save`(*[, validate, check_unique, batch, acl])	Save a document to engine
`set_address`(engine, address_content)	Set address parameter for an Engine
`set_meta_data`(**kwargs)	Set Meta-data of a table
`set_replica_engine`(replica_name, replica_engine)	Set replica engine
`set_version`(doc_id[, version_code])	Set a different version code (uuid) for the given object
`to_db`(*[, catalog, ignore_unknown, engine])	Translate the runtime data into raw data (materialized form of data)
`unlock`([acl])	Unlock the document for update
`update`([_validate, _batch, _acl])	Update existed fields
`validate`()	Validate if all of the component of document follows the predefined rules

Attributes

OPERATORS

action(action_name: str, acl=None, payload: Optional[dict] = None)

Doing an action of a document

Parameters

action_name – The action name
acl – Access Control List
payload – Parameters of action

classmethod analyze(analytic_request: dict, acl=None) → list

Give analyzing results

Parameters

analytic_request – analytic request
acl – User Access Control List

Returns: List of dictionary which holds data

classmethod backup(location: Optional[str] = None, data_encode: Optional[str] = None, data_format: Optional[str] = None, data_store: Optional[str] = None, acl: Optional[Acl] = None, **kwargs)

Backup data of a model

Parameters

acl – User Access List
data_encode (str) – Backup Data Code
data_format (str) – Backup Data Format
data_store (str) – Backup Data Store location
location (str) – Data location to e used by data store
**kwargs – parameter to be passed at engine level

calculate_id() → Optional[str]

Calculate document id from current attributes

Returns: Document id as string if having key field defined else None

classmethod check_acl(acl: Acl, act: str, doc: Optional[Document] = None, raise_error: bool = True)

Check user authorization

Parameters

acl – User’s ACL List
doc – Document to be controlled
act – Action to be performed
raise_error – Raise Error

Notes

when acl is None => Check is deactivated => User has all authorization

classmethod check_dependency(old_data: dict, new_data: Optional[dict] = None, batch: Optional[Batch] = None)

Check if this document supports other document

Parameters

old_data (dict) – Old data to be checked
new_data (dict) – New data to compare
batch (Batch) – Batch data

Returns

Tuple(result, message)

check_scope(scope: list)

Check if data match the data scope

Returns: no exception raise = test passed

classmethod check_update_acl(_acl: Acl, _doc: Optional[Document] = None, **kwargs)

Special ACL check for update. Supporting field group settings

Parameters

_acl – User Access Control
_doc – document to be updated
**kwargs – update parameters

Returns

will raise AuthorizationError when check fails

classmethod collection_action(action_name, acl=None, payload: Optional[dict] = None)

Doing a collection level action

Parameters

action_name – The action name
acl – Access Control List
payload – Parameters of action

classmethod create_collection(acl: Optional[Acl] = None)

Create the collection

Hard way to delete the collection. No cascade relation will be considered

delete(batch: Optional[Batch] = None, acl: Optional[Acl] = None)

Delete a document

Parameters

batch (Batch) – Batch object
acl (Acl) – User Acl list

Returns

deleted object list with collection name, id, operation result

classmethod delete_all(acl: Optional[Acl] = None): Delete every instance of this document type

classmethod dict_to_id(key_values: dict) → str

Calculate document id from key_values

Parameters: key_values (dict) – Key value dictionary
Returns: Document id as string

classmethod dict_to_id_list(key_values: dict) → list

Calculate document id from key_values. Accepting list as value. Output will always be list

Parameters: key_values (dict) – Key value dictionary
Returns: Document id as string

classmethod drop(acl: Optional[Acl] = None)

Drop the collection

Hard way to delete the collection. No cascade relation will be considered

classmethod from_db(_engine: Optional[Type[BaseEngine]] = None, **kwargs)

Transform database data to internal data

Parameters

_engine – engine to be used for convert data
**kwargs (object) – python dict got from system

Returns

python dict object

classmethod from_display(**kwargs)

Transform Display data to internal data

Parameters: **kwargs (object) – display object got from front end
Returns: python dict object

classmethod get_actions()

Get action supported by this document

Returns: Base]
Return type: dictionary[str

classmethod get_address(engine: Optional[Union[Type[BaseEngine], str]] = None)

Get address for an engine

Parameters: engine – Engine class or engine parameter name. default value is the engine attached to the document
Returns: Engine related address parameters

classmethod get_all_fields() → dict: Get All fields of current class

classmethod get_collection_name(engine: Type[BaseEngine] = None)

Get collection name of current class

engine: The collection name is engine dependent

Returns: collection name
Return type: str

get_display_data(lazy: bool = True, catalog: Optional[dict] = None, show_hidden: bool = False)

Get Display data (visualized form of data)

Parameters

lazy (bool) – Left the fields as is if they are not explicitly loaded
catalog (dict) – The catalog of display field in the following format: * {field1: false, field2: false, field3: {sub-field1: true, sub-field2: true}} * field1, field2 is the top-level normal field * field3 is an embedded field/reference field/external field while sub-field1, sub-field2 is its fields * True or false will override the lazy setting, none means using lazy settings
show_hidden (bool) – If hidden value should be shown or not

Returns

python dict object

Notes

We will return the detail form if it is possible

classmethod get_field(parent_object, field_path: str) → Optional[BaseField]

Get field object from a field path

Parameters

parent_object – Parent class object
field_path – field path in format “a.b.c”

Returns: The field object

get_id()

Get document id

Returns: Document ID
Return type: str

classmethod get_meta_data()

Get Meta-data of a table

Returns: Meta data dictionary

get_raw_data()

Get Raw data (materialized form of data)

Returns: python dict object

classmethod get_replica_engines()

Get engines for the keeping replicas

Returns: db_param: Engine class
Return type: replica dictionary

get_runtime_data()

Get Runtime data (calculated from raw data when runtime is activated)

Returns: python dict object

classmethod get_sample(): Get a sample document data

classmethod get_version(doc_id: str) → str

Get a version code (uuid) for the given object

Parameters: doc_id (str) – Document ID
Returns: UUID as version code

classmethod id_to_dict(doc_id: str) → dict

From document id to key

Parameters: doc_id (str) – Document ID
Returns: A dictionary who holds the key fields and the values

classmethod load(*args, _batch: Optional[Batch] = None, _acl: Optional[Acl] = None, **kwargs)

Load a Document from engine

Parameters

_batch (Batch) – Batch object
_acl (Acl) – User Acl to be checked
*args – a list document id (should only have one valid)
**kwargs –

Returns

loaded document instance

lock(timeout: Optional[int] = None, acl: Optional[Acl] = None)

Lock the document for update

Parameters

timeout – Should wait for this amount of time before go timeout
acl (Acl) – User Acl to be checked

classmethod migrate(original_doc: BaseDocument, copy_data: bool = False)

Migrate a document from the original one into new one. Mostly used for data migration between engines

Parameters

original_doc – Original document
copy_data – The new data will be a copy of old one

Returns

The new document with the original reference

Notes:

classmethod objects(*args, _batch: Optional[Batch] = None, _acl: Optional[Acl] = None, _limit: int = 50, _id_only: bool = False, **kwargs)

Search documents

Parameters

_batch (Batch) – Batch object
_acl (Acl) – User ACL to be checked
_limit (int) – Limit result entries length
_id_only (bool) – Only return document ids
**kwargs – Search Configuration

Returns

generator of a found document

Notes

key, str pair: single value search
key, list pair: array_contains_any search
embedded search: a__b means b component of a. a.b means the key’s name is a.b
operators: key is end with __op__. The following op are supported:
- __eq__: Needn’t be added because it is a by default behavior
- __lt__, __le__, __gt__, __ge__, __ne__: as is supposed by the name
- __asc__, __desc__: the result will be ordered by the fields

classmethod purge_version_table(): Purge version table

reload(batch: Optional[Batch] = None, acl: Optional[Acl] = None)

Reload the document from the engine

batch

Batch object

Type: Batch

acl

User ACL to be checked

Type: Acl

Returns: itself with refreshed data

classmethod replicate(task_list: list, acl: Optional[Acl] = None)

Replicate documents

Parameters

task_list – List of dictionary with the following keys: * content: document * op: operation type: “I” for insert, “D” for delete, “U” for update, “L” for load
acl – User Access List

Returns

List of dictionary with the following keys:

id: document id
op: operation type: “I” for insert, “D” for delete, “U” for update, “L” for load
time: time when data is replicated
status: status code of HTTP protocol

Return type

task_results

classmethod restore(location: Optional[str] = None, data_encode: Optional[str] = None, data_format: Optional[str] = None, data_store: Optional[str] = None, acl: Optional[Acl] = None, **kwargs)

Restore data of a model

Parameters

acl – User Access List
data_encode (str) – Backup Data Code
data_format (str) – Backup Data Format
data_store (str) – Backup Data Store location
location (str) – Data location to e used by data store
**kwargs – parameter to be passed at engine level

save(*, validate: bool = True, check_unique: bool = True, batch: Optional[Batch] = None, acl: Optional[Acl] = None)

Save a document to engine

Parameters

validate (bool) – Should the validation to be passed or not
check_unique (bool) – Unique check
batch (Batch) – Batch object
acl (Acl) – User Acl to be checked

Returns

itself (with document id in the case of a new created document)

classmethod set_address(engine: Union[Type[BaseEngine], str], address_content: dict)

Set address parameter for an Engine

Parameters

engine – Engine class or engine parameter name. default value is the engine attached to the document
address_content – Content of address

classmethod set_meta_data(**kwargs)

Set Meta-data of a table

Returns: None

classmethod set_replica_engine(replica_name: str, replica_engine: Type[BaseEngine])

Set replica engine

Parameters

replica_name – name of replica
replica_engine – Replica Engine class

classmethod set_version(doc_id: str, version_code: Optional[str] = None)

Set a different version code (uuid) for the given object

Parameters

doc_id (str) – Document ID
version_code (str) – Version Code to be set. A new UUID if code is not given

Returns

UUID as version code

to_db(*, catalog: Optional[dict] = None, ignore_unknown: bool = False, engine: Optional[Type[BaseEngine]] = None)

Translate the runtime data into raw data (materialized form of data)

Parameters

catalog (dict) – Data catalog to be parsed
ignore_unknown (bool) – When a field is not defined in the document, should we send it back to the database
engine – engine to be used for db data conversion

Returns

python dict object

unlock(acl: Optional[Acl] = None)

Unlock the document for update

Parameters: acl (Acl) – User Acl to be checked

update(_validate=True, _batch: Optional[Batch] = None, _acl: Optional[Acl] = None, **kwargs)

Update existed fields

Args:3: _validate (bool): Should the data be validated before update or not acl (Acl): User ACL to be checked **kwargs: update parameters

Returns: updated document in the form of python object

Notes

Update String Specifications * embedded update: a__b means b component of a. a.b means the key’s name is a.b * operators: key is end with __op__. The following op are supported:

__append__: Append an item to array

__remove__: Remove an item

__delete__: Delete the field

Notes

We need to update the data from all sources

validate()

Validate if all of the component of document follows the predefined rules

Returns: Always None. No exception raised means the validation passed