0.8.0 beta 0

Author: JohnPreston

.pre-commit-config.yaml

@@ -20,6 +20,7 @@ repos:
     hooks:
       - id: pyupgrade
         args: [ "--py38-plus" ]
+        exclude: ^ecs_files_composer/input.py|docs/
 
   - repo: https://github.com/psf/black
     rev: 22.3.0

ecs_files_composer/__init__.py

@@ -1,9 +1,8 @@
-#  -*- coding: utf-8 -*-
 # SPDX-License-Identifier: MPL-2.0
 # Copyright 2020-2021 John Mille<[email protected]>
 
 """Top-level package for ECS Files Composer."""
 
 __author__ = """John Preston"""
 __email__ = "[email protected]"
-__version__ = "0.7.3"
+__version__ = "0.8.0b0"

ecs_files_composer/aws_mgmt.py

@@ -1,4 +1,3 @@
-#  -*- coding: utf-8 -*-
 # SPDX-License-Identifier: MPL-2.0
 # Copyright 2020-2021 John Mille<[email protected]>
 
@@ -67,7 +66,7 @@ def set_session_from_iam_object(iam_config_object, source_session=None):
     return client_session
 
 
-class AwsResourceHandler(object):
+class AwsResourceHandler:
     """
     Class to handle all AWS related credentials init.
     """

ecs_files_composer/cli.py

@@ -1,4 +1,3 @@
-#  -*- coding: utf-8 -*-
 # SPDX-License-Identifier: MPL-2.0
 # Copyright 2020-2021 John Mille<[email protected]>
 

ecs_files_composer/common.py

@@ -1,4 +1,3 @@
-#  -*- coding: utf-8 -*-
 # SPDX-License-Identifier: MPL-2.0
 # Copyright 2020-2021 John Mille<[email protected]>
 

ecs_files_composer/ecs_files_composer.py

@@ -1,4 +1,3 @@
-#  -*- coding: utf-8 -*-
 # SPDX-License-Identifier: MPL-2.0
 # Copyright 2020-2021 John Mille<[email protected]>
 
@@ -75,7 +74,7 @@ def init_config(
     elif secret_config:
         initial_config = {"source": {"Secret": {"SecretId": secret_config}}}
     elif file_path:
-        with open(path.abspath(file_path), "r") as file_fd:
+        with open(path.abspath(file_path)) as file_fd:
             config_content = file_fd.read()
         initial_config = {"content": config_content}
     elif raw:
@@ -97,7 +96,7 @@ def init_config(
     if context:
         initial_config["context"] = context
     start_jobs(jobs_input_def)
-    with open(config_path, "r") as config_fd:
+    with open(config_path) as config_fd:
         try:
             config = yaml.load(config_fd.read(), Loader=Loader)
             LOG.info(f"Successfully loaded YAML config {config_path}")

ecs_files_composer/envsubst.py

@@ -1,4 +1,3 @@
-#  -*- coding: utf-8 -*-
 # SPDX-License-Identifier: MPL-2.0
 # Copyright 2020-2021 John Mille <[email protected]>
 
@@ -14,7 +13,7 @@
 IF_ESCAPED = r"(?<!\\)"
 IF_UNDEFINED = r":-"
 IF_DEFINED = r":+"
-IF_LITTERAL = re.compile("(\$(\{\![^}]+\}))")
+IF_LITTERAL = re.compile(r"(\$(\{\![^}]+\}))")
 
 
 def expandvars(path, default=None, skip_escaped=True, enable_litteral=True):

ecs_files_composer/files_mgmt.py

@@ -1,4 +1,3 @@
-#  -*- coding: utf-8 -*-
 # SPDX-License-Identifier: MPL-2.0
 # Copyright 2020-2021 John Mille<[email protected]>
 
@@ -24,7 +23,7 @@
 from ecs_files_composer.jinja2_filters import JINJA_FILTERS, JINJA_FUNCTIONS
 
 
-class File(FileDef, object):
+class File(FileDef):
     """
     Class to wrap common files actions around
     """
@@ -197,9 +196,8 @@ def set_unix_settings(self):
         try:
             res = subprocess.run(
                 cmd,
-                universal_newlines=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
+                text=True,
+                capture_output=True,
                 shell=False,
             )
         except subprocess.CalledProcessError:
@@ -211,9 +209,8 @@ def set_unix_settings(self):
         try:
             res = subprocess.run(
                 cmd,
-                universal_newlines=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
+                text=True,
+                capture_output=True,
                 shell=False,
             )
         except subprocess.CalledProcessError:
@@ -233,9 +230,8 @@ def exec_post_commands(self):
             try:
                 res = subprocess.run(
                     cmd,
-                    universal_newlines=True,
-                    stdout=subprocess.PIPE,
-                    stderr=subprocess.PIPE,
+                    text=True,
+                    capture_output=True,
                     shell=False,
                 )
             except subprocess.CalledProcessError:

ecs_files_composer/input.py

@@ -1,6 +1,6 @@
 # generated by datamodel-codegen:
 #   filename:  ecs-files-input.json
-#   timestamp: 2022-05-18T06:23:25+00:00
+#   timestamp: 2022-05-18T06:49:26+00:00
 
 from __future__ import annotations
 
@@ -21,67 +21,67 @@ class Context(Enum):
 
 
 class UrlDef(BaseModel):
-    url: AnyUrl | None = Field(None, alias="Url")
-    username: str | None = Field(None, alias="Username")
-    password: str | None = Field(None, alias="Password")
+    url: Optional[AnyUrl] = Field(None, alias="Url")
+    username: Optional[str] = Field(None, alias="Username")
+    password: Optional[str] = Field(None, alias="Password")
 
 
 class IamOverrideDef(BaseModel):
-    role_arn: str | None = Field(None, alias="RoleArn")
-    session_name: str | None = Field(
+    role_arn: Optional[str] = Field(None, alias="RoleArn")
+    session_name: Optional[str] = Field(
         "S3File@EcsConfigComposer",
         alias="SessionName",
         description="Name of the IAM session",
     )
-    external_id: str | None = Field(
+    external_id: Optional[str] = Field(
         None,
         alias="ExternalId",
         description="The External ID to use when using sts:AssumeRole",
     )
-    region_name: str | None = Field(None, alias="RegionName")
-    access_key_id: str | None = Field(
+    region_name: Optional[str] = Field(None, alias="RegionName")
+    access_key_id: Optional[str] = Field(
         None, alias="AccessKeyId", description="AWS Access Key Id to use for session"
     )
-    secret_access_key: str | None = Field(
+    secret_access_key: Optional[str] = Field(
         None, alias="SecretAccessKey", description="AWS Secret Key to use for session"
     )
-    session_token: str | None = Field(None, alias="SessionToken")
+    session_token: Optional[str] = Field(None, alias="SessionToken")
 
 
 class CommandsDef(BaseModel):
-    __root__: list[str] = Field(..., description="List of commands to run")
+    __root__: List[str] = Field(..., description="List of commands to run")
 
 
 class X509CertDef(BaseModel):
     class Config:
         extra = Extra.allow
 
-    dir_path: str | None = None
-    email_address: EmailStr | None = Field(
+    dir_path: Optional[str] = None
+    email_address: Optional[EmailStr] = Field(
         "[email protected]", alias="emailAddress"
     )
-    common_name: None | (
+    common_name: Optional[
         constr(
             regex=r"^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])\.)*([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]{0,61}[A-Za-z0-9])\Z"
         )
-    ) = Field(None, alias="commonName")
-    country_name: str | None = Field("AW", alias="countryName", regex="^[A-Z]+$")
-    locality_name: str | None = Field("AWS", alias="localityName")
-    state_or_province_name: str | None = Field("AWS", alias="stateOrProvinceName")
-    organization_name: str | None = Field("AWS", alias="organizationName")
-    organization_unit_name: str | None = Field("AWS", alias="organizationUnitName")
-    validity_end_in_seconds: float | None = Field(
+    ] = Field(None, alias="commonName")
+    country_name: Optional[str] = Field("AW", alias="countryName", regex="^[A-Z]+$")
+    locality_name: Optional[str] = Field("AWS", alias="localityName")
+    state_or_province_name: Optional[str] = Field("AWS", alias="stateOrProvinceName")
+    organization_name: Optional[str] = Field("AWS", alias="organizationName")
+    organization_unit_name: Optional[str] = Field("AWS", alias="organizationUnitName")
+    validity_end_in_seconds: Optional[float] = Field(
         8035200,
         alias="validityEndInSeconds",
         description="Validity before cert expires, in seconds. Default 3*31*24*60*60=3Months",
     )
     key_file_name: str = Field(..., alias="keyFileName")
     cert_file_name: str = Field(..., alias="certFileName")
-    group: str | None = Field(
+    group: Optional[str] = Field(
         "root",
         description="UNIX group name or GID owner of the file. Default to root(0)",
     )
-    owner: str | None = Field(
+    owner: Optional[str] = Field(
         "root", description="UNIX user or UID owner of the file. Default to root(0)"
     )
 
@@ -90,76 +90,78 @@ class Certificates(BaseModel):
     class Config:
         extra = Extra.forbid
 
-    x509: dict[str, X509CertDef] | None = None
+    x509: Optional[Dict[str, X509CertDef]] = None
 
 
 class Commands(BaseModel):
-    post: CommandsDef | None = Field(
+    post: Optional[CommandsDef] = Field(
         None, description="Commands to run after the file was retrieved"
     )
-    pre: CommandsDef | None = Field(
+    pre: Optional[CommandsDef] = Field(
         None,
         description="Commands executed prior to the file being fetched, after `depends_on` completed",
     )
 
 
 class SsmDef(BaseModel):
-    parameter_name: str | None = Field(None, alias="ParameterName")
-    iam_override: IamOverrideDef | None = Field(None, alias="IamOverride")
+    parameter_name: Optional[str] = Field(None, alias="ParameterName")
+    iam_override: Optional[IamOverrideDef] = Field(None, alias="IamOverride")
 
 
 class SecretDef(BaseModel):
     secret_id: str = Field(..., alias="SecretId")
-    version_id: str | None = Field(None, alias="VersionId")
-    version_stage: str | None = Field(None, alias="VersionStage")
-    iam_override: IamOverrideDef | None = Field(None, alias="IamOverride")
+    version_id: Optional[str] = Field(None, alias="VersionId")
+    version_stage: Optional[str] = Field(None, alias="VersionStage")
+    iam_override: Optional[IamOverrideDef] = Field(None, alias="IamOverride")
 
 
 class S3Def(BaseModel):
     bucket_name: str = Field(
         ..., alias="BucketName", description="Name of the S3 Bucket"
     )
-    bucket_region: str | None = Field(
+    bucket_region: Optional[str] = Field(
         None,
         alias="BucketRegion",
         description="S3 Region to use. Default will ignore or retrieve via s3:GetBucketLocation",
     )
     key: str = Field(..., alias="Key", description="Full path to the file to retrieve")
-    iam_override: IamOverrideDef | None = Field(None, alias="IamOverride")
+    iam_override: Optional[IamOverrideDef] = Field(None, alias="IamOverride")
 
 
 class SourceDef(BaseModel):
-    url: UrlDef | None = Field(None, alias="Url")
-    ssm: SsmDef | None = Field(None, alias="Ssm")
-    s3: S3Def | None = Field(None, alias="S3")
-    secret: SecretDef | None = Field(None, alias="Secret")
+    url: Optional[UrlDef] = Field(None, alias="Url")
+    ssm: Optional[SsmDef] = Field(None, alias="Ssm")
+    s3: Optional[S3Def] = Field(None, alias="S3")
+    secret: Optional[SecretDef] = Field(None, alias="Secret")
 
 
 class FileDef(BaseModel):
     class Config:
         extra = Extra.allow
 
-    path: str | None = None
-    content: str | None = Field(None, description="The raw content of the file to use")
-    source: SourceDef | None = None
-    encoding: Encoding | None = "plain"
-    group: str | None = Field(
+    path: Optional[str] = None
+    content: Optional[str] = Field(
+        None, description="The raw content of the file to use"
+    )
+    source: Optional[SourceDef] = None
+    encoding: Optional[Encoding] = "plain"
+    group: Optional[str] = Field(
         "root",
         description="UNIX group name or GID owner of the file. Default to root(0)",
     )
-    owner: str | None = Field(
+    owner: Optional[str] = Field(
         "root", description="UNIX user or UID owner of the file. Default to root(0)"
     )
-    mode: str | None = Field("0644", description="UNIX file mode")
-    context: Context | None = "plain"
-    ignore_if_failed: bool | None = Field(
+    mode: Optional[str] = Field("0644", description="UNIX file mode")
+    context: Optional[Context] = "plain"
+    ignore_if_failed: Optional[bool] = Field(
         False,
         description="Whether or not the failure to retrieve the file should stop the execution",
     )
-    commands: Commands | None = None
+    commands: Optional[Commands] = None
 
 
 class Model(BaseModel):
-    files: dict[str, FileDef]
-    certificates: Certificates | None = None
-    iam_override: IamOverrideDef | None = Field(None, alias="IamOverride")
+    files: Dict[str, FileDef]
+    certificates: Optional[Certificates] = None
+    iam_override: Optional[IamOverrideDef] = Field(None, alias="IamOverride")

ecs_files_composer/jinja2_filters/__init__.py

@@ -1,4 +1,3 @@
-#  -*- coding: utf-8 -*-
 # SPDX-License-Identifier: MPL-2.0
 # Copyright 2020-2021 John Mille<[email protected]>
 
@@ -32,9 +31,7 @@ def define_metadata(for_task=False):
     elif environ.get(meta_v4, None):
         meta_url = environ.get(meta_v4)
     else:
-        raise EnvironmentError(
-            "No ECS Metadata URL provided. This filter only works on ECS"
-        )
+        raise OSError("No ECS Metadata URL provided. This filter only works on ECS")
     if for_task:
         return requests.get(f"{meta_url}/task")
     else: