OAuth Authenticator

For plugins that require OAuth, OAuthAuthenticator provides an easy interface to login to thirds party services. By implementing You can nest an inherited version class of OAuthAuthenticator in your Plugin item and call authenticator.authenticate().

YourOauthAuthenticator class should implement:

  • get_oauth_url() -> str to setup OAuth endpoint
  • get_tokens_from_code() -> Dict[str, str] to retrieve tokens from the service with returned OAuth code
  • refresh_tokens() -> None to get new token pairs

Example

Lets build an example oauth authenticator, to show its class structure

class ExampleOAuthAuthenticator(OAuthAuthenticator):

    def get_oauth_url(self):
        return "https://example.com/oauth"

    def get_tokens_from_code(self, code):
        return {
            'access_token': 'dummy_access_token',
            'refresh_token': 'dummy_refresh_token'
        }

    def refresh_tokens(self, refreshToken):
        return {
            'access_token': 'refreshed_dummy_access_token',
            'refresh_token': 'refreshed_dummy_refresh_token'
        }

    def verify_access_token(self, token):
        if token:
            return True
    
    def present_url_to_user(self, *args):
        # NORMALLY YOU WOULD NOT IMPLEMENT THIS
        # mocking user interaction
        self.pluginRun.status = RUN_USER_ACTION_COMPLETED
        self.pluginRun.account[0].code = "dummy code"
        self.pluginRun.account[0].update(self.client)
        self.pluginRun.update(self.client)
class MyOAuthPlugin(PluginBase):

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def run(self):
        print("running")
        self.login()
        print("Login completed!")
        
    def login(self):
        auth = ExampleOAuthAuthenticator(self.client, self.pluginRun)
        auth.authenticate()
        print(f"logged in with {auth.get_account().accessToken}")

    def add_to_schema(self):
        self.client.add_to_schema(MyItem("my name", 10))
client = PodClient()

# Create a dummy account to use for authentication within the plugin
account = Account(service="my_plugin_service", identifier="username", secret="password")
# Create a run to enable plugin runtime
run = PluginRun("pymemri", "pymemri.plugin.pluginbase", "MyOAuthPlugin")
run.add_edge('account', account)
account.update(client)
run.update(client)
plugin = MyOAuthPlugin(pluginRun=run, client=client)
plugin.run()
# from pymemri.plugin.schema import Account
# from pymemri.plugin.pluginbase import PluginRun, run_plugin_from_run_id

# client = PodClient()

# # Create a dummy account to use for authentication within the plugin
# account = Account(service="my_plugin_service", identifier="username", secret="password")
# client.create(account)

# # Create a run to enable plugin runtime
# run = PluginRun("pymemri", "pymemri.plugin.pluginbase", "MyOAuthPlugin")
# run.add_edge('account', account)
# client.create(run)
# client.update(account)

# plugin = run_plugin_from_run_id(run.id, client)

# # check if authentication worked
# assert plugin.pluginRun.account[0].identifier == "username"
# assert plugin.pluginRun.account[0].accessToken == "dummy_access_token"

# # set a state
# plugin.pluginRun.status = "test state"
# plugin.pluginRun.update(client)
# plugin.pluginRun = client.get(plugin.pluginRun.id)
# assert plugin.pluginRun.status == "test state"