kubernetes-client/python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: use f-string

Browse Source
This commit is contained in:
Ananya Nayak 2023-09-13 23:24:16 +05:30
parent 397d06b832
commit 298725de92
16 changed files with 42 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
examples/annotate_deployment.py
View File

@ -62,8 +62,7 @@ def main():
time.sleep(1)
before_annotating = apps_v1_api.read_namespaced_deployment(
'deploy-nginx', 'default')
print('Before annotating, annotations: %s' %
before_annotating.metadata.annotations)
print(f"Before annotating, annotations: {before_annotating.metadata.annotations}")
annotations = [
{
@ -80,8 +79,7 @@ def main():
time.sleep(1)
after_annotating = apps_v1_api.read_namespaced_deployment(
name='deploy-nginx', namespace='default')
print('After annotating, annotations: %s' %
after_annotating.metadata.annotations)
print(f"After annotating, annotations: {after_annotating.metadata.annotations}")
if __name__ == "__main__":

5
examples/api_discovery.py
View File

@ -27,8 +27,7 @@ def main():
config.load_kube_config()
print("Supported APIs (* is preferred version):")
print("%-40s %s" %
("core", ",".join(client.CoreApi().get_api_versions().versions)))
print(f"{'core':<40} {','.join(client.CoreApi().get_api_versions().versions)}")
for api in client.ApisApi().get_api_versions().groups:
versions = []
for v in api.versions:
@ -38,7 +37,7 @@ def main():
name += "*"
name += v.version
versions.append(name)
print("%-40s %s" % (api.name, ",".join(versions)))
print(f"{api.name:<40} {','.join(versions)}")
if __name__ == '__main__':

2
examples/apply_from_directory.py
View File

@ -1,4 +1,4 @@
from kubernetes import client,config,utils
from kubernetes import client, config, utils
def main():
config.load_kube_config()

4
examples/apply_from_single_file.py
View File

@ -1,9 +1,9 @@
from kubernetes import client,config,utils
from kubernetes import client, config, utils
def main():
config.load_kube_config()
k8s_client = client.ApiClient()
yaml_file = 'examples/configmap-demo-pod.yml'
yaml_file = 'examples/yaml_dir/configmap-demo-pod.yml'
utils.create_from_yaml(k8s_client,yaml_file,verbose=True)
if __name__ == "__main__":

15
examples/cluster_scoped_custom_object.py
View File

@ -100,10 +100,7 @@ def main():
plural="crontabs",
)
print("%s\t\t%s" % ("NAME", "CRON-SPEC"))
print(
"%s\t%s\n" %
(resource["metadata"]["name"],
resource["spec"]["cronSpec"]))
print(f"{resource['metadata']['name']}\t{resource['spec']['cronSpec']}\n")
# patch the `spec.cronSpec` field of the custom resource
patched_resource = api.patch_cluster_custom_object(
@ -115,10 +112,7 @@ def main():
)
print("[INFO] Custom resource `test-crontab` patched to update the cronSpec schedule!\n")
print("%s\t\t%s" % ("NAME", "PATCHED-CRON-SPEC"))
print(
"%s\t%s\n" %
(patched_resource["metadata"]["name"],
patched_resource["spec"]["cronSpec"]))
print(f"{patched_resource['metadata']['name']}\t{patched_resource['spec']['cronSpec']}\n")
# patch the `metadata.labels` field of the custom resource
patched_resource = api.patch_cluster_custom_object(
@ -130,10 +124,7 @@ def main():
)
print("[INFO] Custom resource `test-crontab` patched to apply new metadata labels!\n")
print("%s\t\t%s" % ("NAME", "PATCHED_LABELS"))
print(
"%s\t%s\n" %
(patched_resource["metadata"]["name"],
patched_resource["metadata"]["labels"]))
print(f"{patched_resource['metadata']['name']}\t{patched_resource['metadata']['labels']}\n")
# delete the custom resource "test-crontab"
api.delete_cluster_custom_object(

2
examples/deployment_create.py
View File

@ -34,7 +34,7 @@ def main():
k8s_apps_v1 = client.AppsV1Api()
resp = k8s_apps_v1.create_namespaced_deployment(
body=dep, namespace="default")
print("Deployment created. status='%s'" % resp.metadata.name)
print(f"Deployment created. Status='{resp.metadata.name}'")
if __name__ == '__main__':

3
examples/in_cluster_config.py
View File

@ -58,8 +58,7 @@ def main():
print("Listing pods with their IPs:")
ret = v1.list_pod_for_all_namespaces(watch=False)
for i in ret.items:
print("%s\t%s\t%s" %
(i.status.pod_ip, i.metadata.namespace, i.metadata.name))
print(f"{i.status.pod_ip}\t{i.metadata.namespace}\t{i.metadata.name}")
if __name__ == '__main__':

10
examples/job_crud.py
View File

@ -27,7 +27,7 @@ JOB_NAME = "pi"
def create_job_object():
# Configureate Pod template container
# Configure Pod template container
container = client.V1Container(
name="pi",
image="perl",
@ -54,7 +54,7 @@ def create_job(api_instance, job):
api_response = api_instance.create_namespaced_job(
body=job,
namespace="default")
print("Job created. status='%s'" % str(api_response.status))
print(f"Job created. status='{str(api_response.status)}'")
get_job_status(api_instance)
@ -68,7 +68,7 @@ def get_job_status(api_instance):
api_response.status.failed is not None:
job_completed = True
sleep(1)
print("Job status='%s'" % str(api_response.status))
print(f"Job status='{str(api_response.status)}'")
def update_job(api_instance, job):
@ -78,7 +78,7 @@ def update_job(api_instance, job):
name=JOB_NAME,
namespace="default",
body=job)
print("Job updated. status='%s'" % str(api_response.status))
print(f"Job updated. status='{str(api_response.status)}'")
def delete_job(api_instance):
@ -88,7 +88,7 @@ def delete_job(api_instance):
body=client.V1DeleteOptions(
propagation_policy='Foreground',
grace_period_seconds=5))
print("Job deleted. status='%s'" % str(api_response.status))
print(f"Job deleted. status='{str(api_response.status)}'")
def main():

8
examples/multiple_clusters.py
View File

@ -43,13 +43,11 @@ def main():
print("\nList of pods on %s:" % cluster1)
for i in client1.list_pod_for_all_namespaces().items:
print("%s\t%s\t%s" %
(i.status.pod_ip, i.metadata.namespace, i.metadata.name))
print(f"{i.status.pod_ip}\t{i.metadata.namespace}\t{i.metadata.name}")
print("\n\nList of pods on %s:" % cluster2)
print(f"\n\nList of pods on {cluster2}:")
for i in client2.list_pod_for_all_namespaces().items:
print("%s\t%s\t%s" %
(i.status.pod_ip, i.metadata.namespace, i.metadata.name))
print(f"{i.status.pod_ip}\t{i.metadata.namespace}\t{i.metadata.name}")
if __name__ == '__main__':

2
examples/node_labels.py
View File

@ -44,7 +44,7 @@ def main():
# Patching the node labels
for node in node_list.items:
api_response = api_instance.patch_node(node.metadata.name, body)
print("%s\t%s" % (node.metadata.name, node.metadata.labels))
print(f"{node.metadata.name}\t{node.metadata.labels}")
if __name__ == '__main__':

5
examples/out_of_cluster_config.py
View File

@ -13,7 +13,7 @@
# limitations under the License.
"""
Shows how to load a Kubernetes config from outside of the cluster.
Shows how to load a Kubernetes config from outside the cluster.
"""
from kubernetes import client, config
@ -29,8 +29,7 @@ def main():
print("Listing pods with their IPs:")
ret = v1.list_pod_for_all_namespaces(watch=False)
for i in ret.items:
print("%s\t%s\t%s" %
(i.status.pod_ip, i.metadata.namespace, i.metadata.name))
print(f"{i.status.pod_ip}\t{i.metadata.namespace}\t{i.metadata.name}")
if __name__ == '__main__':

2
examples/pick_kube_config_context.py
View File

@ -37,7 +37,7 @@ def main():
# utility
config.load_kube_config(context=option)
print("Active host is %s" % configuration.Configuration().host)
print(f"Active host is {configuration.Configuration().host}")
v1 = client.CoreV1Api()
print("Listing pods with their IPs:")

2
examples/pod_config_list.py
View File

@ -38,7 +38,7 @@ def main():
# utility
config.load_kube_config(context=option)
print("Active host is %s" % configuration.Configuration().host)
print(f"Active host is {configuration.Configuration().host}")
v1 = client.CoreV1Api()
print("Listing pods with their IPs:")

14
examples/pod_exec.py
View File

@ -33,11 +33,11 @@ def exec_commands(api_instance):
namespace='default')
except ApiException as e:
if e.status != 404:
print("Unknown error: %s" % e)
print(f"Unknown error: {e}")
exit(1)
if not resp:
print("Pod %s does not exist. Creating it..." % name)
print(f"Pod {name} does not exist. Creating it...")
pod_manifest = {
'apiVersion': 'v1',
'kind': 'Pod',
@ -98,22 +98,22 @@ def exec_commands(api_instance):
while resp.is_open():
resp.update(timeout=1)
if resp.peek_stdout():
print("STDOUT: %s" % resp.read_stdout())
print(f"STDOUT: {resp.read_stdout()}")
if resp.peek_stderr():
print("STDERR: %s" % resp.read_stderr())
print(f"STDERR: {resp.read_stderr()}")
if commands:
c = commands.pop(0)
print("Running command... %s\n" % c)
print(f"Running command... {c}\n")
resp.write_stdin(c + "\n")
else:
break
resp.write_stdin("date\n")
sdate = resp.readline_stdout(timeout=3)
print("Server date command returns: %s" % sdate)
print(f"Server date command returns: {sdate}")
resp.write_stdin("whoami\n")
user = resp.readline_stdout(timeout=3)
print("Server user is: %s" % user)
print(f"Server user is: {user}")
resp.close()

19
examples/pod_portforward.py
View File

@ -66,11 +66,11 @@ def portforward_commands(api_instance):
namespace='default')
except ApiException as e:
if e.status != 404:
print("Unknown error: %s" % e)
print(f"Unknown error: {e}")
exit(1)
if not resp:
print("Pod %s does not exist. Creating it..." % name)
print(f"Pod {name} does not exist. Creating it...")
pod_manifest = {
'apiVersion': 'v1',
'kind': 'Pod',
@ -119,7 +119,7 @@ def portforward_commands(api_instance):
if error is None:
print("No port forward errors on port 80.")
else:
print("Port 80 has the following error: %s" % error)
print(f"Port 80 has the following error: {error}")
# Monkey patch socket.create_connection which is used by http.client and
# urllib.request. The same can be done with urllib3.util.connection.create_connection
@ -147,10 +147,10 @@ def portforward_commands(api_instance):
break
else:
raise RuntimeError(
"Unable to find service port: %s" % port)
f"Unable to find service port: {port}")
label_selector = []
for key, value in service.spec.selector.items():
label_selector.append("%s=%s" % (key, value))
label_selector.append(f"{key}={value}")
pods = api_instance.list_namespaced_pod(
namespace, label_selector=",".join(label_selector)
)
@ -168,11 +168,10 @@ def portforward_commands(api_instance):
break
else:
raise RuntimeError(
"Unable to find service port name: %s" % port)
f"Unable to find service port name: {port}")
elif dns_name[1] != 'pod':
raise RuntimeError(
"Unsupported resource type: %s" %
dns_name[1])
f"Unsupported resource type: {dns_name[1]}")
pf = portforward(api_instance.connect_get_namespaced_pod_portforward,
name, namespace, ports=str(port))
return pf.socket(port)
@ -181,10 +180,10 @@ def portforward_commands(api_instance):
# Access the nginx http server using the
# "<pod-name>.pod.<namespace>.kubernetes" dns name.
response = urllib_request.urlopen(
'http://%s.pod.default.kubernetes' % name)
f'http://{name}.pod.default.kubernetes')
html = response.read().decode('utf-8')
response.close()
print('Status Code: %s' % response.code)
print(f'Status Code: {response.code}')
print(html)

3
examples/remote_cluster.py
View File

@ -52,8 +52,7 @@ def main():
print("Listing pods with their IPs:")
ret = v1.list_pod_for_all_namespaces(watch=False)
for i in ret.items:
print("%s\t%s\t%s" %
(i.status.pod_ip, i.metadata.namespace, i.metadata.name))
print(f"{i.status.pod_ip}\t{i.metadata.namespace}\t{i.metadata.name}")
if __name__ == '__main__':