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 committed by Yu Liao
parent d77daf461c
commit 8a7e24b9cf
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__':