Setting the file. One moment.
Common · Hf Cloud Sagemaker Production Defaults · huggingface/skills · Skills Docs
Back to the top of the page Next
Script Deploy Async
scripts/ _common.py
Python · 148 lines · 5 KB
14
from
typing
import
Any
15
16 from botocore.exceptions import ClientError
17
18
19 # Shared tagging convention. CreatedBy lets us find every resource these
20 # scripts have ever created via:
21 # aws resourcegroupstaggingapi get-resources \
22 # --tag-filters Key=CreatedBy,Values=agentic-deploy-skills
23 CREATED_BY_TAG_VALUE = "agentic-deploy-skills"
24
25
26 def log (prefix: str , msg: str ) -> None :
27 """Stream a log line to stderr with a per-script prefix."""
28 print ( f "[ { prefix } ] { msg } " , file = sys.stderr, flush = True )
29
30
31 def parse_env (env_args: list[ str ]) -> dict[ str , str ]:
32 """Parse --env KEY=VALUE flags from argparse into a dict."""
33 env: dict[ str , str ] = {}
34 for item in env_args:
35 if "=" not in item:
36 raise SystemExit ( f "--env must be KEY=VALUE, got: { item } " )
37 k, v = item.split( "=" , 1 )
38 env[k] = v
39 return env
40
41
42 def make_endpoint_name (model_name: str , override: str | None ) -> str :
43 """Generate a SageMaker-legal endpoint name.
44
45 Default format: <model-name>-<YYYYMMDD-HHMM>. SageMaker names are limited
46 to 63 chars and must be DNS-friendly (lowercase, hyphens, no underscores).
47 """
48 if override:
49 return override
50 stamp = datetime.now(timezone.utc).strftime( "%Y%m %d -%H%M" )
51 base = model_name.replace( "_" , "-" ).lower()
52 return f " { base } - { stamp } " [: 63 ]
53
54
55 def build_tags (
56 * ,
57 project: str ,
58 caller_arn: str ,
59 environment: str ,
60 model_s3_uri: str | None = None ,
61 extra: dict[ str , str ] | None = None ,
62 ) -> list[dict[ str , str ]]:
63 """Build the shared tag set applied to every resource we create.
64
65 `extra` lets the caller add deployment-mode-specific tags (e.g.
66 `{"InferenceMode": "async"}`) without forking the schema.
67 """
68 owner = caller_arn.split( "/" )[ - 1 ] if "/" in caller_arn else caller_arn
69 tags = [
70 { "Key" : "Project" , "Value" : project},
71 { "Key" : "Owner" , "Value" : owner},
72 { "Key" : "Environment" , "Value" : environment},
73 { "Key" : "CreatedBy" , "Value" : CREATED_BY_TAG_VALUE },
74 ]
75 if model_s3_uri:
76 tags.append({ "Key" : "ModelArtifact" , "Value" : model_s3_uri})
77 if extra:
78 for k, v in extra.items():
79 tags.append({ "Key" : k, "Value" : v})
80 return tags
81
82
83 def create_model (
84 sm: Any,
85 * ,
86 model_name: str ,
87 image_uri: str ,
88 role_arn: str ,
89 model_s3_uri: str | None ,
90 env: dict[ str , str ],
91 tags: list[dict[ str , str ]],
92 log_prefix: str = "deploy" ,
93 ) -> str :
94 """Create a SageMaker Model. Idempotent — reuses on AlreadyExists.
95
96 Returns the model name on success. The model definition itself is mode-
97 independent: same call, same parameters, whether the eventual endpoint
98 is real-time or async.
99 """
100 log(log_prefix, f "Creating model: { model_name } " )
101 primary_container: dict[ str , Any] = { "Image" : image_uri}
102 if env:
103 primary_container[ "Environment" ] = env
104 if model_s3_uri:
105 primary_container[ "ModelDataUrl" ] = model_s3_uri
106
107 try :
108 sm.create_model(
109 ModelName = model_name,
110 PrimaryContainer = primary_container,
111 ExecutionRoleArn = role_arn,
112 Tags = tags,
113 )
114 except ClientError as e:
115 if "Cannot create already existing model" in str (e):
116 log(log_prefix, f "Model { model_name } already exists — reusing" )
117 else :
118 raise
119 return model_name
120
121
122 def wait_for_endpoint (
123 sm: Any,
124 endpoint_name: str ,
125 timeout_minutes: int = 30 ,
126 log_prefix: str = "deploy" ,
127 ) -> None :
128 """Poll DescribeEndpoint until InService, or raise on Failed/timeout."""
129 log(log_prefix, f "Waiting for { endpoint_name } to reach InService (up to { timeout_minutes } min)..." )
130 start = time.time()
131 deadline = start + (timeout_minutes * 60 )
132
133 while time.time() < deadline:
134 resp = sm.describe_endpoint( EndpointName = endpoint_name)
135 status = resp[ "EndpointStatus" ]
136 elapsed = int (time.time() - start)
137
138 if status == "InService" :
139 log(log_prefix, f "InService after { elapsed } s" )
140 return
141 if status == "Failed" :
142 reason = resp.get( "FailureReason" , "(no reason given)" )
143 raise RuntimeError ( f "Endpoint creation failed after { elapsed } s: { reason } " )
144
145 log(log_prefix, f " status= { status } elapsed= { elapsed } s" )
146 time.sleep( 30 )
147
148 raise TimeoutError ( f "Endpoint did not reach InService within { timeout_minutes } minutes" )