FM / subiquitycore

/snap/core18/2999/usr/share/subiquity/subiquitycore/

/snap/core18/2999/usr/share/subiquity/subiquitycore

Name	Size	Mode	Actions
controllers/	-	0755	rm
models/	-	0755	rm
ui/	-	0755	rm
controller.py	2607	0644	edit dl rm
core.py	5519	0644	edit dl rm
log.py	1568	0644	edit dl rm
palette.py	3337	0644	edit dl rm
prober.py	2353	0644	edit dl rm
signals.py	4625	0644	edit dl rm
snapd.py	1517	0644	edit dl rm
user.py	2376	0644	edit dl rm
utils.py	5778	0644	edit dl rm
view.py	1567	0644	edit dl rm
__init__.py	736	0644	edit dl rm

Edit: /snap/core18/2999/usr/share/subiquity/subiquitycore/snapd.py (1517B)

import json
import logging
from urllib.parse import (
    quote_plus,
    urlencode,
    )

import requests_unixsocket

log = logging.getLogger('subiquitycore.snapd')

# Every method in this module blocks. Do not call them from the main thread!

class SnapdConnection:
    def __init__(self, root, sock):
        self.root = root
        self.url_base = "http+unix://{}/".format(quote_plus(sock))
        self.session = requests_unixsocket.Session()

def get(self, path, **args):
        if args:
            path += '?' + urlencode(args)
        return self.session.get(self.url_base + path, timeout=60)

def post(self, path, body, **args):
        if args:
            path += '?' + urlencode(args)
        return self.session.post(
            self.url_base + path, data=json.dumps(body),
            timeout=60)

# Copyright 2019 Canonical, Ltd. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import json import logging from urllib.parse import ( quote_plus, urlencode, ) import requests_unixsocket log = logging.getLogger('subiquitycore.snapd') # Every method in this module blocks. Do not call them from the main thread! class SnapdConnection: def __init__(self, root, sock): self.root = root self.url_base = "http+unix://{}/".format(quote_plus(sock)) self.session = requests_unixsocket.Session() def get(self, path, **args): if args: path += '?' + urlencode(args) return self.session.get(self.url_base + path, timeout=60) def post(self, path, body, **args): if args: path += '?' + urlencode(args) return self.session.post( self.url_base + path, data=json.dumps(body), timeout=60)