This example demonstrates how to build a microservice using the Pip.Services framework in Python, implementing a 'HelloFriend' service with a REST controller and pluggable persistence layers (MySQL and PostgreSQL). The architecture follows a Controller-Service-Persistence pattern where the specific database implementation is determined at runtime via configuration/environment variables.
from pip_services4_data.data import IStringIdentifiable
from pip_services4_http.controller import RestController
from pip_services4_components.config import IConfigurable
from pip_services4_components.refer import IReferences, IReferenceable, Descriptor
from pip_services4_components.build import Factory
from pip_services4_container.container import ProcessContainer
from pip_services4_http.build import DefaultRpcFactory
from pip_services4_mysql.persistence import IdentifiableMySqlPersistence
from pip_services4_postgres.persistence import IdentifiablePostgresPersistence
from pip_services4_data.query import FilterParams
from typing import Optional
from abc import ABC
import bottle
# 1. Data Model
class MyFriend(IStringIdentifiable):
def __init__(self, id: str, type: str, name: str):
self.id = id
self.type = type
self.name = name
# 2. Persistence Interface
class IMyDataPersistence(ABC):
def get_one_random(self, trace_id: Optional[str], filter: FilterParams) -> MyFriend:
raise NotImplemented()
def create(self, trace_id: Optional[str], item: MyFriend) -> MyFriend:
raise NotImplemented()
# 3. MySQL Implementation
class HelloFriendPersistence1(IdentifiableMySqlPersistence, IMyDataPersistence):
def __init__(self):
super(HelloFriendPersistence1, self).__init__('myfriends3')
def _define_schema(self):
self._clear_schema()
self._ensure_schema(f"CREATE TABLE IF NOT EXISTS `{self._table_name}` (id VARCHAR(32) PRIMARY KEY, `type` VARCHAR(50), `name` TEXT)")
def _compose_filter(self, filter: FilterParams):
filter = filter or FilterParams()
type = filter.get_as_nullable_string('type')
name = filter.get_as_nullable_string('name')
condition = ''
if type: condition += f"`type`='{type}'"
if name: condition += f"`name`='{name}'"
return condition
def get_one_random(self, trace_id: str, filter: FilterParams) -> MyFriend:
return super().get_one_random(trace_id, self._compose_filter(filter))
# 4. Service Logic
class HelloFriendService(IConfigurable, IReferenceable):
def __init__(self):
self.__persistence = None
def set_references(self, references: IReferences):
self.__persistence = references.get_one_required(Descriptor("hello-friend", "persistence", "*", "*", "1.0"))
def greeting(self):
filter_param = FilterParams.from_tuples("type", "friend")
selected_friend = self.__persistence.get_one_random(None, filter_param)
return f"Hello, {selected_friend.name} !"
def create(self, trace_id: Optional[str], item: MyFriend) -> MyFriend:
return self.__persistence.create(trace_id, item)
# 5. REST Controller
class HelloFriendRestController(RestController):
def __init__(self):
super().__init__()
self._base_route = "/hello_friend"
self._service = None
def set_references(self, references):
self._service = references.get_one_required(Descriptor('hello-friend', 'service', 'default', '*', '1.0'))
def register(self):
self.register_route(method="GET", route="/greeting", handler=self.greeting)
self.register_route(method="GET", route="/greeting_create", handler=self.create)
def greeting(self):
return self.send_result(self._service.greeting())
def create(self):
item = MyFriend(bottle.request.query["id"], bottle.request.query["type"], bottle.request.query["name"])
return self.send_result(self._service.create(self._get_trace_id(), item))
# 6. Factory and Process Container
class HelloFriendControllerFactory(Factory):
def __init__(self):
super().__init__()
self.register_as_type(Descriptor('hello-friend', 'controller', 'http', '*', '1.0'), HelloFriendRestController)
self.register_as_type(Descriptor('hello-friend', 'service', 'default', '*', '1.0'), HelloFriendService)
self.register_as_type(Descriptor('hello-friend', 'persistence', 'mysql', '*', '1.0'), HelloFriendPersistence1)
self.register_as_type(Descriptor('hello-friend', 'persistence', 'postgres', '*', '1.0'), HelloFriendPersistence2)
class HelloFriendProcess(ProcessContainer):
def __init__(self):
super().__init__('hello-friend', 'HelloFriend microservice')
self._config_path = './configDTC.yaml'
self._factories.add(HelloFriendControllerFactory())
self._factories.add(DefaultRpcFactory())
if __name__ == '__main__':
import os
os.environ["MYSQL_ENABLED"] = "True" # Toggle between MySQL and Postgres
runner = HelloFriendProcess()
runner.run()
