First implementation

twtxt_registry_client/__init__.py

@@ -0,0 +1 @@
+from twtxt_registry_client.client import RegistryClient

twtxt_registry_client/__main__.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+from urllib.parse import urlsplit, urlunsplit
+import click
+from twtxt_registry_client import RegistryClient
+
+
+@click.group(name='twtxt-registry')
+@click.argument('registry_url', required=True)
+@click.pass_context
+def cli(ctx, registry_url):
+    scheme, netloc, path, query, fragment = urlsplit(registry_url)
+    if not scheme:
+        scheme = 'https'
+    if not netloc and path:
+        netloc, _, path = path.partition('/')
+    registry_url = urlunsplit((scheme, netloc, path, query, fragment))
+    ctx.obj = RegistryClient(registry_url)
+
+
+@cli.command()
+@click.argument('nickname', required=True)
+@click.argument('url', required=True)
+@click.pass_context
+def register(ctx, nickname, url):
+    # TODO: Use twtxt's config to guess the user's info
+    click.echo(ctx.obj.register(nickname, url).text)
+
+
+@cli.command()
+@click.option('-q', '--query')
+@click.pass_context
+def users(ctx, query):
+    click.echo(ctx.obj.list_users(q=query).text)
+
+
+@cli.command()
+@click.option('-q', '--query')
+@click.pass_context
+def tweets(ctx, query):
+    click.echo(ctx.obj.list_tweets(q=query).text)
+
+
+@cli.command()
+@click.argument('url', required=True)
+@click.pass_context
+def mentions(ctx, url):
+    # TODO: Allow names instead of URLs for known users
+    # TODO: Use twtxt's config to default to the user's name
+    click.echo(ctx.obj.list_mentions(url).text)
+
+
+@cli.command()
+@click.argument('name', required=True)
+@click.pass_context
+def tag(ctx, name):
+    click.echo(ctx.obj.list_tag_tweets(name).text)
+
+
+if __name__ == '__main__':
+    cli()

twtxt_registry_client/client.py

@@ -0,0 +1,58 @@
+import urllib
+import requests
+
+
+class RegistryClient(object):
+
+    def __init__(self, registry_url):
+        self.registry_url = registry_url
+
+    def request(self, method, endpoint, *, format='plain', **params):
+        resp = method(
+            '/'.join([self.registry_url, format, endpoint]),
+            # Ignore parameters with None values
+            params={k: v for k, v in params.items() if v},
+        )
+        resp.raise_for_status()
+        return resp
+
+    def get(self, *args, **kwargs):
+        return self.request(requests.get, *args, **kwargs)
+
+    def post(self, *args, **kwargs):
+        return self.request(requests.post, *args, **kwargs)
+
+    def register(self, nickname, url, *, format='plain'):
+        return self.post(
+            'users',
+            format=format,
+            nickname=nickname,
+            url=url,
+        )
+
+    def list_users(self, *, q=None, format='plain'):
+        return self.get(
+            'users',
+            q=q,
+            format=format,
+        )
+
+    def list_tweets(self, *, q=None, format='plain'):
+        return self.get(
+            'tweets',
+            q=q,
+            format=format,
+        )
+
+    def list_mentions(self, url, *, format='plain'):
+        return self.get(
+            'mentions',
+            url=url,
+            format=format,
+        )
+
+    def list_tag_tweets(self, name, *, format='plain'):
+        return self.get(
+            'tags/{}'.format(urllib.quote(name)),
+            format=format,
+        )