Skip to content
Snippets Groups Projects

Добавлен общий код PerxisDataProvider, GrpcChannel и моделей

Merged Podosochnyy Maxim requested to merge feature/add-shared-code-of-data-providers into master
3 files
+ 285
0
Compare changes
  • Side-by-side
  • Inline
Files
3
perxis/channel.py 0 → 100644
+ 57
0
from typing import Optional
import grpc
from perxis.auth import APIKeyPlugin
from perxis.models import PerxisEnviron
class GrpcChannel:
call_credentials: grpc.CallCredentials
channel_credentials: grpc.ChannelCredentials
composite_credentials: grpc.ChannelCredentials
channel: Optional[grpc.Channel] = None
def __init__(
self,
target: str,
metadata_plugin: grpc.AuthMetadataPlugin,
project_name: str,
project_version: str,
call_credentials_name: str = "api-key",
) -> None:
self.target = target
self.metadata_plugin = metadata_plugin
self.call_credentials_name = call_credentials_name
self._init_credentials()
self.connect(project_name, project_version)
def _init_credentials(self) -> None:
self.call_credentials = grpc.metadata_call_credentials(
self.metadata_plugin, name=self.call_credentials_name
)
self.channel_credentials = grpc.ssl_channel_credentials()
self.composite_credentials = grpc.composite_channel_credentials(
self.channel_credentials, self.call_credentials
)
def connect(self, project_name: str, project_version: str) -> None:
if not self.channel:
self.channel = grpc.aio.secure_channel(
self.target, self.composite_credentials,
options=(
("grpc.primary_user_agent", f"{project_name}/{project_version}"),
)
)
async def close(self) -> None:
await self.channel.close()
def get_grpc_channel(perxis_environ: PerxisEnviron, project_name: str, project_version: str) -> GrpcChannel:
return GrpcChannel(
target=perxis_environ.target,
metadata_plugin=APIKeyPlugin(token=perxis_environ.api_key),
project_name=project_name,
project_version=project_version
)
Loading