1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
# coding: utf-8
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
"""
FILE: sample_set_get_image.py
DESCRIPTION:
This sample demonstrates setting and getting OCI and non-OCI images to a repository.
USAGE:
python sample_set_get_image.py
Set the environment variables with your own values before running the sample:
1) CONTAINERREGISTRY_ENDPOINT - The URL of your Container Registry account
"""
import os
import json
from io import BytesIO
from dotenv import find_dotenv, load_dotenv
from azure.containerregistry import (
ContainerRegistryClient,
DigestValidationError,
)
from utilities import get_authority, get_credential
class SetGetImage(object):
def __init__(self):
load_dotenv(find_dotenv())
self.endpoint = os.environ["CONTAINERREGISTRY_ENDPOINT"]
self.authority = get_authority(self.endpoint)
self.credential = get_credential(self.authority)
def set_oci_image(self):
# [START upload_blob_and_manifest]
self.repository_name = "sample-oci-image"
layer = BytesIO(b"Sample layer")
config = BytesIO(
json.dumps(
{
"sample config": "content",
}
).encode()
)
with ContainerRegistryClient(self.endpoint, self.credential) as client:
# Upload a layer
layer_digest, layer_size = client.upload_blob(self.repository_name, layer)
print(f"Uploaded layer: digest - {layer_digest}, size - {layer_size}")
# Upload a config
config_digest, config_size = client.upload_blob(self.repository_name, config)
print(f"Uploaded config: digest - {config_digest}, size - {config_size}")
# Create an oci image with config and layer info
oci_manifest = {
"config": {
"mediaType": "application/vnd.oci.image.config.v1+json",
"digest": config_digest,
"sizeInBytes": config_size,
},
"schemaVersion": 2,
"layers": [
{
"mediaType": "application/vnd.oci.image.layer.v1.tar",
"digest": layer_digest,
"size": layer_size,
"annotations": {
"org.opencontainers.image.ref.name": "artifact.txt",
},
},
],
}
# Set the image with tag "latest"
manifest_digest = client.set_manifest(self.repository_name, oci_manifest, tag="latest")
print(f"Uploaded manifest: digest - {manifest_digest}")
# [END upload_blob_and_manifest]
def get_oci_image(self):
# [START download_blob_and_manifest]
with ContainerRegistryClient(self.endpoint, self.credential) as client:
# Get the image
get_manifest_result = client.get_manifest(self.repository_name, "latest")
received_manifest = get_manifest_result.manifest
print(f"Got manifest:\n{received_manifest}")
# Download and write out the layers
for layer in received_manifest["layers"]:
# Remove the "sha256:" prefix from digest
layer_file_name = layer["digest"].split(":")[1]
try:
stream = client.download_blob(self.repository_name, layer["digest"])
with open(layer_file_name, "wb") as layer_file:
for chunk in stream:
layer_file.write(chunk)
except DigestValidationError:
print(f"Downloaded layer digest value did not match. Deleting file {layer_file_name}.")
os.remove(layer_file_name)
print(f"Got layer: {layer_file_name}")
# Download and write out the config
config_file_name = "config.json"
try:
stream = client.download_blob(self.repository_name, received_manifest["config"]["digest"])
with open(config_file_name, "wb") as config_file:
for chunk in stream:
config_file.write(chunk)
except DigestValidationError:
print(f"Downloaded config digest value did not match. Deleting file {config_file_name}.")
os.remove(config_file_name)
print(f"Got config: {config_file_name}")
# [END download_blob_and_manifest]
def delete_blob(self):
# [START delete_blob]
with ContainerRegistryClient(self.endpoint, self.credential) as client:
get_manifest_result = client.get_manifest(self.repository_name, "latest")
received_manifest = get_manifest_result.manifest
# Delete the layers
for layer in received_manifest["layers"]:
client.delete_blob(self.repository_name, layer["digest"])
# Delete the config
client.delete_blob(self.repository_name, received_manifest["config"]["digest"])
# [END delete_blob]
def delete_oci_image(self):
# [START delete_manifest]
with ContainerRegistryClient(self.endpoint, self.credential) as client:
get_manifest_result = client.get_manifest(self.repository_name, "latest")
# Delete the image
client.delete_manifest(self.repository_name, get_manifest_result.digest)
# [END delete_manifest]
def set_get_oci_image(self):
self.set_oci_image()
self.get_oci_image()
self.delete_blob()
self.delete_oci_image()
def set_get_docker_image(self):
repository_name = "sample-docker-image"
with ContainerRegistryClient(self.endpoint, self.credential) as client:
# Upload a layer
layer = BytesIO(b"Sample layer")
layer_digest, layer_size = client.upload_blob(repository_name, layer)
print(f"Uploaded layer: digest - {layer_digest}, size - {layer_size}")
# Upload a config
config = BytesIO(json.dumps({"sample config": "content"}).encode())
config_digest, config_size = client.upload_blob(repository_name, config)
print(f"Uploaded config: digest - {config_digest}, size - {config_size}")
# create a Docker image object in Docker v2 Manifest format
docker_manifest = {
"config": {
"digest": config_digest,
"mediaType": "application/vnd.docker.container.image.v1+json",
"size": config_size,
},
"layers": [
{
"digest": layer_digest,
"mediaType": "application/vnd.docker.image.rootfs.diff.tar.gzip",
"size": layer_size,
}
],
"mediaType": "application/vnd.docker.distribution.manifest.v2+json",
"schemaVersion": 2,
}
# Set the image with one custom media type
client.set_manifest(
repository_name, docker_manifest, tag="sample", media_type=str(docker_manifest["mediaType"])
)
# Get the image
get_manifest_result = client.get_manifest(repository_name, "sample")
received_manifest = get_manifest_result.manifest
print(received_manifest)
received_manifest_media_type = get_manifest_result.media_type
print(received_manifest_media_type)
# Delete the image
client.delete_manifest(repository_name, get_manifest_result.digest)
if __name__ == "__main__":
sample = SetGetImage()
sample.set_get_oci_image()
sample.set_get_docker_image()
|
