Setting the file. One moment.
Mirror Image · Hf Cloud Serving Image Selection · huggingface/skills · Skills Docs
Back to the top of the page scripts/ mirror_image.py
Python · 145 lines · 5 KB
17 """
18
19 from __future__ import annotations
20
21 import os
22 import shutil
23 import subprocess
24 import sys
25
26
27 def log (msg: str ) -> None :
28 print ( f "[mirror_image] { msg } " , file = sys.stderr, flush = True )
29
30
31 def require (cmd: str ) -> str :
32 exe = shutil.which(cmd)
33 if not exe:
34 log( f "ERROR: { cmd } not installed or not on PATH" )
35 sys.exit( 1 )
36 return exe
37
38
39 def run (args: list[ str ], ** kwargs) -> subprocess.CompletedProcess:
40 return subprocess.run(args, capture_output = True , text = True , ** kwargs)
41
42
43 def resolve_region () -> str :
44 for var in ( "AWS_REGION" , "AWS_DEFAULT_REGION" ):
45 if os.environ.get(var):
46 return os.environ[var]
47 proc = run([shutil.which( "aws" ), "configure" , "get" , "region" ])
48 return proc.stdout.strip() if proc.returncode == 0 else ""
49
50
51 def main () -> int :
52 if len (sys.argv) < 3 :
53 log( f "Usage: { os.path.basename(sys.argv[ 0 ]) } <public-image-uri> <private-repo-name> [<tag-override>]" )
54 return 64
55
56 public_uri = sys.argv[ 1 ]
57 private_repo = sys.argv[ 2 ]
58 tag_override = sys.argv[ 3 ] if len (sys.argv) > 3 else ""
59
60 aws = require( "aws" )
61 docker = require( "docker" )
62 if run([docker, "info" ]).returncode != 0 :
63 log( "ERROR: docker daemon not running" )
64 return 1
65
66 # Extract tag
67 if tag_override:
68 tag = tag_override
69 elif ":" in public_uri.rsplit( "/" , 1 )[ - 1 ]:
70 tag = public_uri.rsplit( ":" , 1 )[ - 1 ]
71 else :
72 log( "ERROR: public URI has no tag — refusing implicit ':latest'" )
73 return 1
74
75 acct = run([aws, "sts" , "get-caller-identity" , "--query" , "Account" , "--output" , "text" ])
76 if acct.returncode != 0 :
77 log( "ERROR: 'aws sts get-caller-identity' failed. Configure AWS credentials." )
78 return 1
79 account_id = acct.stdout.strip()
80
81 region = resolve_region()
82 if not region:
83 log( "ERROR: no AWS region. Set AWS_REGION or configure profile." )
84 return 1
85
86 registry = f " { account_id } .dkr.ecr. { region } .amazonaws.com"
87 private_uri = f " { registry } / { private_repo } : { tag } "
88
89 log( f "Public : { public_uri } " )
90 log( f "Private: { private_uri } " )
91
92 reg = [ "--region" , region]
93
94 # Create private repo if missing
95 if run([aws, "ecr" , "describe-repositories" , "--repository-names" , private_repo, * reg]).returncode != 0 :
96 log( f "Creating private ECR repo: { private_repo } " )
97 run(
98 [aws, "ecr" , "create-repository" , "--repository-name" , private_repo,
99 "--image-scanning-configuration" , "scanOnPush=true" , * reg]
100 )
101
102 # Skip if tag already exists in private
103 if run(
104 [aws, "ecr" , "describe-images" , "--repository-name" , private_repo,
105 "--image-ids" , f "imageTag= { tag } " , * reg]
106 ).returncode == 0 :
107 log( f "Tag ' { tag } ' already in private ECR — skipping pull/push" )
108 print (private_uri)
109 return 0
110
111 # Auth. ECR Public auth always uses us-east-1.
112 def docker_login (password: str , registry_host: str ) -> bool :
113 login = subprocess.run(
114 [docker, "login" , "--username" , "AWS" , "--password-stdin" , registry_host],
115 input = password, capture_output = True , text = True ,
116 )
117 return login.returncode == 0
118
119 pub_pw = run([aws, "ecr-public" , "get-login-password" , "--region" , "us-east-1" ])
120 if pub_pw.returncode != 0 or not docker_login(pub_pw.stdout, "public.ecr.aws" ):
121 log( "ERROR: failed to authenticate to public.ecr.aws" )
122 return 1
123
124 priv_pw = run([aws, "ecr" , "get-login-password" , * reg])
125 if priv_pw.returncode != 0 or not docker_login(priv_pw.stdout, registry):
126 log( f "ERROR: failed to authenticate to { registry } " )
127 return 1
128
129 log( f "Pulling { public_uri } (may take several minutes)..." )
130 if subprocess.run([docker, "pull" , public_uri]).returncode != 0 :
131 log( "ERROR: docker pull failed" )
132 return 1
133 subprocess.run([docker, "tag" , public_uri, private_uri])
134 log( "Pushing to private ECR..." )
135 if subprocess.run([docker, "push" , private_uri]).returncode != 0 :
136 log( "ERROR: docker push failed" )
137 return 1
138
139 log( "Done." )
140 print (private_uri)
141 return 0
142
143
144 if __name__ == "__main__" :
145 sys.exit(main())