'os' imported but unused (F401)

'typing.Optional' imported but unused (F401)

'src.services' imported but unused (F401)

ingest_url: str = Field(description="URL of ingest API") raster_url: str = Field(description="URL of raster API") stac_url: str = Field(description="URL of STAC API") mwaa_env: str = Field(description="MWAA URL") - class Config(): + class Config: env_file = ".env"

def get_username(claims: security.HTTPBasicCredentials = Depends(decode_token)): return claims["sub"] -def get_and_validate_token(token: security.HTTPAuthorizationCredentials = Depends(token_scheme)): + +def get_and_validate_token( + token: security.HTTPAuthorizationCredentials = Depends(token_scheme), +): decode_token(token) return token + def _get_secret_hash(username: str, client_id: str, client_secret: str) -> str: # A keyed-hash message authentication code (HMAC) calculated using # the secret key of a user pool client and username plus the client # ID in the message.

@workflows_app.post( "/dataset/publish", tags=["Dataset"], dependencies=[Depends(auth.get_username)] ) async def publish_dataset( - token = Depends(auth.get_and_validate_token), + token=Depends(auth.get_and_validate_token), dataset: Union[schemas.ZarrDataset, schemas.COGDataset] = Body( ..., discriminator="data_type" ), ): # Construct and load collection

collection = collection.model_dump(by_alias=True) url = f"{ingest_api}/collections" headers = { "Authorization": f"Bearer {token}", - "Content-Type": "application/json" + "Content-Type": "application/json", } response = requests.post(url, json=collection, headers=headers) if response.status_code == 200: print("Success:", response.json()) else: print("Error:", response.status_code, response.text) + # TODO refactor class Publisher: common_fields = [ "title",

collection = collection.model_dump(by_alias=True) url = f"{ingest_api}/collections" headers = { "Authorization": f"Bearer {token}", - "Content-Type": "application/json" + "Content-Type": "application/json", } response = requests.post(url, json=collection, headers=headers) if response.status_code == 200: print("Success:", response.json()) else: print("Error:", response.status_code, response.text) -

Linting failures detected. See check runs with annotations for details.

'os' imported but unused (F401)

'typing.Optional' imported but unused (F401)

ingest_url: str = Field(description="URL of ingest API") raster_url: str = Field(description="URL of raster API") stac_url: str = Field(description="URL of STAC API") mwaa_env: str = Field(description="MWAA URL") - class Config(): + class Config: env_file = ".env"

'src.services' imported but unused (F401)

def get_username(claims: security.HTTPBasicCredentials = Depends(decode_token)): return claims["sub"] -def get_and_validate_token(token: security.HTTPAuthorizationCredentials = Depends(token_scheme)): + +def get_and_validate_token( + token: security.HTTPAuthorizationCredentials = Depends(token_scheme), +): decode_token(token) return token + def _get_secret_hash(username: str, client_id: str, client_secret: str) -> str: # A keyed-hash message authentication code (HMAC) calculated using # the secret key of a user pool client and username plus the client # ID in the message.

@workflows_app.post( "/dataset/publish", tags=["Dataset"], dependencies=[Depends(auth.get_username)] ) async def publish_dataset( - token = Depends(auth.get_and_validate_token), + token=Depends(auth.get_and_validate_token), dataset: Union[schemas.ZarrDataset, schemas.COGDataset] = Body( ..., discriminator="data_type" ), ): # Construct and load collection

collection = collection.model_dump(by_alias=True) url = f"{ingest_api}/collections" headers = { "Authorization": f"Bearer {token}", - "Content-Type": "application/json" + "Content-Type": "application/json", } response = requests.post(url, json=collection, headers=headers) if response.status_code == 200: print("Success:", response.json()) else: print("Error:", response.status_code, response.text) + # TODO refactor class Publisher: common_fields = [ "title",

collection = collection.model_dump(by_alias=True) url = f"{ingest_api}/collections" headers = { "Authorization": f"Bearer {token}", - "Content-Type": "application/json" + "Content-Type": "application/json", } response = requests.post(url, json=collection, headers=headers) if response.status_code == 200: print("Success:", response.json()) else: print("Error:", response.status_code, response.text) -

Node.js 16 actions are deprecated. Please update the following actions to use Node.js 20: actions/checkout@v2, actions/setup-python@v4, wearerequired/lint-action@v2. For more information see: https://github.blog/changelog/2023-09-22-github-actions-transitioning-from-node-16-to-node-20/.