Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add support for channel notices #710

Closed
2 changes: 1 addition & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

We [keep a changelog.](http://keepachangelog.com/)

## 1.12.3
## 1.12.3 - 2024-02-22

### Tickets closed

Expand Down
2 changes: 1 addition & 1 deletion binstar_client/__about__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,4 +4,4 @@

__all__ = ['__version__']

__version__ = '1.12.2'
__version__ = '1.12.3'
20 changes: 20 additions & 0 deletions binstar_client/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -624,3 +624,23 @@ def user_licenses(self):
res = self.session.get(url)
self._check_response(res)
return res.json()

def create_notices(self, login, label, notices):
"""Create channel notices"""
url = f'{self.domain}/channels/{login}/{label}/notices'
res = self.session.post(url, json=notices)
self._check_response(res)

def remove_notices(self, login, label):
"""Remove channel notices"""
url = f'{self.domain}/channels/{login}/{label}/notices'
res = self.session.delete(url)
self._check_response(res)

def notices(self, login, label):
"""Retrieve channel notices"""
url = f'{self.domain}/channels/{login}/{label}/notices'
res = self.session.get(url)
self._check_response(res)

return res.json()
121 changes: 121 additions & 0 deletions binstar_client/commands/notices.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
"""
Anaconda channel notices utilities
"""
import argparse
import json
import logging
import pathlib
import typing

from binstar_client import errors
from binstar_client.utils import get_server_api, get_config

logger = logging.getLogger('binstar.notices')


def main(args):
"""Entry point for notices command"""
api = get_server_api(
token=args.token, site=args.site, config=get_config(args.site)
)
api.check_server()

if args.user is None:
args.user = api.user().get('login')

if args.user is None:
message: str = 'Unable to determine owner in user; please make sure you are logged in'
logger.error(message)
raise errors.BinstarError(message)

if args.notices:
api.create_notices(args.user, args.label, args.notices)

elif args.remove:
api.remove_notices(args.user, args.label)

elif args.get:
notices = api.notices(args.user, args.label)
logger.info(json.dumps(notices, indent=2))


class NoticesAction(argparse.Action):
"""
Used to parse the notices argument as either an JSON string or JSON file
"""

def __call__(self, parser, namespace, values, *args, **kwargs):
"""
We first test if ``values`` is a file and then try to parse it as JSON.
If it isn't, we assume it's a JSON string itself and attempt to parse it.
"""
try:
path = pathlib.Path(values)
except TypeError as error:
message: str = 'Notices argument must be defined as a string'
logger.error(message)
raise SystemExit(1) from error

if path.exists():
try:
with path.open(encoding='utf-8') as file_pointer:
values = file_pointer.read()
except OSError as error:
message: str = f'Unable to read provided JSON file: {error}'
logger.error(message)
raise SystemExit(1) from error

try:
data = json.loads(values)
except json.JSONDecodeError as error:
message: str = 'Unable to parse provided JSON; please make sure it is valid JSON'
logger.error(message)
raise SystemExit(1) from error

setattr(namespace, self.dest, data)


def add_parser(subparsers: typing.Any) -> None:
"""
Set options for notices subcommand
"""
description: str = 'Create, modify and delete channels notices in your Anaconda repository'
parser: argparse.ArgumentParser = subparsers.add_parser(
'notices',
formatter_class=argparse.RawDescriptionHelpFormatter,
help=description, description=description,
epilog=__doc__,
)

parser.add_argument(
'-l', '--label',
default='main',
help='Label to use for channel notice'
)

parser.add_argument(
'-u', '--user',
help='User account or Organization, defaults to the current user',
)

agroup = parser.add_argument_group('actions')
group = agroup.add_mutually_exclusive_group()
group.add_argument(
'--create',
dest='notices',
metavar='notices',
action=NoticesAction,
help='Create notices; existing notices will be replaced'
)
group.add_argument(
'--remove',
action='store_true',
help='Remove notices'
)
group.add_argument(
'--get',
action='store_true',
help='Display notices'
)

parser.set_defaults(main=main)
Loading
Loading