Synapse structure

The custom synapse consists of the main.py and requirements.txt files. First includes the Synapse logic and the later includes all dependancies (native python dependancies don't need to be specified). At minimum the requirement file needs to include the syncari-sdk library.

Structuring a Synapse

When creating a new synapse we recommend to import all available assets from Syncari client, synapse and models. Bellow is a boilerplate example of the import:

from syncari.rest.client import *
from syncari.synapse.abstract_synapse import *
from syncari.models import *

After we define the import of the modules we need to set the execute method, which serves as the entry point for our platform to execute the Synapse's methods.

def execute(request):
    return YourSynapseName(request.data).execute()

After the execute method we need to add an instance of the Synapse class. The name of the instance should match the name used in the execute method.

YourSynapseName(Synapse):
    ...

Inside the class we can define Synapse's functionality using the following available methods: synapse_info, test, describe, refresh_token, get_access_token, read, get_by_id, create, update, delete, extract_webhook_identifier and process_webhook.

Synapse_info, test and describe methods are always mandatory to be present, as they are responsible for configuring UI elements, testing connection and configuring the synapse's schema. 

Other methods are not necessary to have always implemented, but it is mandatory to have them present in the Synapse class. When a method is not used or is not supported for a specific entity, the synapse should return a NotSupportedException to the platform:

def method_name(self):
    raise NotSupportedException(ErrorResponse(status_code=405, message=method_name not supported.))

Example Synapse Code

Bellow you can see the example structure of a synapse, you can reuse as your starting boilerplate. It includes all the needed libraries, a single entity seeded schema, required methods and the __init__ method. 

__init__ method should always include super().__init__(request) It is recommended to also initiate an instance of the Syncari REST inside the method using self.client = SyncariRestClient(self.connection.endpoint, {'api_token':self.connection.authConfig.accessToken}) or self.client = SyncariRestClient(self.connection.endpoint, self.connection.authConfig).

When there is a use case for multiple rest clients, for example when it's needed to pull data from different APIs, those additional clients can be added here as well. Additionally when using different clients like SOAP or SFTP one, those can be also defined here.

Example main.py

from syncari.rest.client import *
from syncari.synapse.abstract_synapse import *
from syncari.models import *

entity_schemas = {
    'contact': {
        'apiName':'contact',
        'displayName':'Contact',
        'attributes':[
            {'apiName':'id', 'dataType':'string', 'isIdField':True, 'displayName':'Id'},
            {'apiName':'firstName', 'dataType':'string', 'displayName':'First Name'},
            {'apiName':'lastName', 'dataType':'string', 'displayName':'Last Name'},
            {'apiName':'companyName', 'dataType':'string', 'displayName':'Company Name'},
            {'apiName':'email', 'dataType':'string', 'displayName':'Email'},
            {'apiName':'dateAdded', 'dataType':'datetime', 'displayName':'Created At','isCreatedAtField':True,'isSystem':True},
            {'apiName':'dateUpdated', 'dataType':'datetime', 'displayName':'Updated At','isUpdatedAtField':True, 'isWatermarkField':True,'isSystem':True}
        ]
     }
}

def execute(request):
    return YourSynapseName(request.data).execute()

class YourSynapseName(Synapse):

    def __init__(self, request: Request) -> None:
        super().__init__(request)
        self.client = SyncariRestClient(self.connection.authConfig.endpoint, self.connection.authConfig)

    def synapse_info(self) -> SynapseInfo:
        return SynapseInfo(
            name='yoursynapsename', category='category',
            metadata=UIMetadata(displayName='Your Synapse'),
            supportedAuthTypes=[
                AuthMetadata(
                    authType=AuthType.API_KEY,
                    label='API Key',
                    fields=[
                        AuthField(name='token',
                                  label='API Key',
                                  dataType=DataType.PASSWORD)
                        ])],
            configuredFields=[AuthField(name='endpoint', label='Endpoint URL', dataType=DataType.STRING)])

    def test(self, connection: Connection) -> Connection:
        return connection

    def describe(self, desc_request: DescribeRequest) -> List[Schema]:
        entities = desc_request.entities
        if entities is None or not entities:
            entities = entity_schemas.keys()
        return [Schema.parse_obj(entity_schemas[entity]) for entity in entities if entity in entity_schemas]

    def read(self, sync_request: SyncRequest) -> ReadResponse:
        raise NotSupportedException(ErrorResponse(status_code=405, message="read not supported."))

    def get_by_id(self, sync_request: SyncRequest) -> List[Record]:
        raise NotSupportedException(ErrorResponse(status_code=405, message="get_by_id not supported."))

    def create(self, sync_request: SyncRequest) -> List[Result]:
        raise NotSupportedException(ErrorResponse(status_code=405, message="create not supported."))

    def update(self, sync_request: SyncRequest) -> List[Result]:
        raise NotSupportedException(ErrorResponse(status_code=405, message="update not supported."))

    def delete(self, sync_request: SyncRequest) -> List[Result]:
        raise NotSupportedException(ErrorResponse(status_code=405, message="delete not supported."))

    def extract_webhook_identifier(self, webhook_request: WebhookRequest) -> str:
        raise NotSupportedException(ErrorResponse(status_code=405, message="extract_webhook_identifier not supported."))

    def process_webhook(self, webhook_request: WebhookRequest) -> List[Record]:
        raise NotSupportedException(ErrorResponse(status_code=405, message="process_webhook not supported."))
  

Example requirements.txt

syncari-sdk