refactor: extract client

This commit is contained in:
Marc 2022-10-30 12:01:46 -04:00
parent 662d5fa00a
commit c6abb508a3
2 changed files with 79 additions and 72 deletions

View file

@ -0,0 +1,76 @@
import urllib.request
import urllib.parse
import typing
import logging
import json
logger = logging.getLogger(__name__)
class SlackClient:
_token: str
def __init__(self, token: str):
self._token = token
@property
def headers(self):
return {
"Authorization": f"Bearer {self._token}",
}
def _post(self, url: str, payload):
request = urllib.request.Request(
url,
urllib.parse.urlencode(payload).encode(),
method="POST",
headers=self.headers,
)
response = urllib.request.urlopen(request)
response_status = response.status
response_data = response.read()
logger.debug("API request: %s", str(payload))
logger.debug("API response: %s", str(response_data))
if response_status != 200:
raise Exception("Failed due to an API error.")
response_data = json.loads(response_data)
if not response_data["ok"]:
raise Exception("Failed due to an API error.")
def update_status(
self,
status: str,
emoticon: typing.Optional[str] = None,
expiration: typing.Optional[int] = None,
):
"""
Sets the Slack status of the given user to <status>, optionally with <emoticon> if provided.
If an expiration is provided, the status is set to expire after this time.
Reference: https://api.slack.com/methods/users.profile.set
"""
payload = {
"profile": {
"status_text": status,
"status_emoji": emoticon or "",
"status_expiration": expiration or 0,
}
}
self._post("https://slack.com/api/users.profile.set", payload)
def set_do_not_disturb(self, duration_minutes: int):
"""
Silences notifications, potentially with the specified duration.
Reference: https://api.slack.com/methods/dnd.setSnooze
"""
payload = {"num_minutes": duration_minutes}
self._post("https://slack.com/api/dnd.setSnooze", payload)

View file

@ -13,8 +13,6 @@ With preset:
SLACK_TOKEN=XXX slack-status-cli set --preset <preset-name> --duration <duration_description> SLACK_TOKEN=XXX slack-status-cli set --preset <preset-name> --duration <duration_description>
""" """
import urllib.request
import urllib.parse
import typing import typing
import os import os
import logging import logging
@ -26,6 +24,8 @@ import json
import pathlib import pathlib
import sys import sys
import client as slack_client
# Debug mode modifies the log level used for reporting. If truthy, # Debug mode modifies the log level used for reporting. If truthy,
# extra information is included in each run to diagnose common # extra information is included in each run to diagnose common
# issues. # issues.
@ -61,75 +61,6 @@ Configuration = collections.namedtuple(
) )
class SlackClient:
_token: str
def __init__(self, token: str):
self._token = token
@property
def headers(self):
return {
"Authorization": f"Bearer {self._token}",
}
def _post(self, url: str, payload):
request = urllib.request.Request(
url,
urllib.parse.urlencode(payload).encode(),
method="POST",
headers=self.headers,
)
response = urllib.request.urlopen(request)
response_status = response.status
response_data = response.read()
logger.debug("API request: %s", str(payload))
logger.debug("API response: %s", str(response_data))
if response_status != 200:
raise Exception("Failed due to an API error.")
response_data = json.loads(response_data)
if not response_data["ok"]:
raise Exception("Failed due to an API error.")
def update_status(
self,
status: str,
emoticon: typing.Optional[str] = None,
expiration: typing.Optional[int] = None,
):
"""
Sets the Slack status of the given user to <status>, optionally with <emoticon> if provided.
If an expiration is provided, the status is set to expire after this time.
Reference: https://api.slack.com/methods/users.profile.set
"""
payload = {
"profile": {
"status_text": status,
"status_emoji": emoticon or "",
"status_expiration": expiration or 0,
}
}
self._post("https://slack.com/api/users.profile.set", payload)
def set_do_not_disturb(self, duration_minutes: int):
"""
Silences notifications, potentially with the specified duration.
Reference: https://api.slack.com/methods/dnd.setSnooze
"""
payload = {"num_minutes": duration_minutes}
self._post("https://slack.com/api/dnd.setSnooze", payload)
def parse_input(known_presets: typing.List[str]) -> ParsedUserInput: def parse_input(known_presets: typing.List[str]) -> ParsedUserInput:
""" """
Handles command-line argument parsing and help text display. Handles command-line argument parsing and help text display.
@ -247,7 +178,7 @@ def run():
if not token: if not token:
raise Exception("Slack token not provided.") raise Exception("Slack token not provided.")
client = SlackClient(token=token) client = slack_client.SlackClient(token=token)
status_text = args.text status_text = args.text
status_icon = args.icon status_icon = args.icon