76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
import json
|
|
from urllib.parse import urljoin
|
|
|
|
import requests as RQ
|
|
|
|
|
|
class Portainer(object):
|
|
def __init__(self, url, username, passwd):
|
|
self.url = url
|
|
self.session = RQ.Session()
|
|
jwt = self.session.post(
|
|
urljoin(self.url, "api/auth"),
|
|
json={"username": username, "password": passwd},
|
|
).json()
|
|
self.session.headers.update(
|
|
{"Authorization": "Bearer {0[jwt]}".format(jwt)})
|
|
|
|
def containers(self, container_id=None):
|
|
if container_id is None:
|
|
res = self.session.get(
|
|
urljoin(self.url, "api/endpoints/1/docker/containers/json"),
|
|
params={
|
|
"all": 1,
|
|
"filters": json.dumps(
|
|
{"label": ["com.docker.compose.project=tvstack"]}
|
|
),
|
|
},
|
|
)
|
|
else:
|
|
res = self.session.get(
|
|
urljoin(
|
|
self.url,
|
|
"api/endpoints/1/docker/containers/{}/json".format(container_id),
|
|
))
|
|
res.raise_for_status()
|
|
res = res.json()
|
|
if container_id is None:
|
|
for container in res:
|
|
pass
|
|
# print("Gettings stats for",container['Id'])
|
|
# container['stats']=self.stats(container['Id'])
|
|
# container['top']=self.top(container['Id'])
|
|
else:
|
|
res["stats"] = self.stats(container_id)
|
|
res["top"] = self.top(container_id)
|
|
return res
|
|
|
|
def top(self, container_id):
|
|
res = self.session.get(
|
|
urljoin(
|
|
self.url,
|
|
"api/endpoints/1/docker/containers/{}/top".format(container_id),
|
|
))
|
|
res.raise_for_status()
|
|
res = res.json()
|
|
cols = res["Titles"]
|
|
ret = []
|
|
|
|
return res
|
|
|
|
def stats(self, container_id):
|
|
res = self.session.get(
|
|
urljoin(
|
|
self.url,
|
|
"api/endpoints/1/docker/containers/{}/stats".format(container_id),
|
|
),
|
|
params={
|
|
"stream": False},
|
|
)
|
|
res.raise_for_status()
|
|
return res.json()
|
|
|
|
def test(self):
|
|
self.containers()
|
|
return {}
|