ogdc_runner.service.auth_routes module¶

Module containing FastAPI routes requiring an access token.

class ogdc_runner.service.auth_routes.CreateUserResponse(**data: Any) → None¶

Bases: BaseModel

Parameters:

data (Any)
message (str)

message: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ogdc_runner.service.auth_routes.OutputResponse(**data: Any) → None¶

Bases: BaseModel

Parameters:

data (Any)
data_url (str)

data_url: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ogdc_runner.service.auth_routes.StatusResponse(**data: Any) → None¶

Bases: BaseModel

Parameters:

data (Any)
recipe_workflow_name (str)
status (str | None)
timestamp (datetime)

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

recipe_workflow_name: str¶

status: str | None¶

timestamp: datetime¶

class ogdc_runner.service.auth_routes.SubmitRecipeRequest(**data: Any) → None¶

Bases: BaseModel

Parameters:

data (Any)
recipe_path (str)
overwrite (bool)

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

overwrite: bool¶

recipe_path: str¶

class ogdc_runner.service.auth_routes.SubmitRecipeResponse(**data: Any) → None¶

Bases: BaseModel

Parameters:

data (Any)
message (str)
recipe_workflow_name (str | None)

message: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

recipe_workflow_name: str | None¶

ogdc_runner.service.auth_routes.create_user_route(form_data: Annotated[fastapi.security.OAuth2PasswordRequestForm, fastapi.Depends], session: Annotated[Session, fastapi.Depends], authenticated_user: Annotated[User, fastapi.Depends]) → CreateUserResponse¶

Create a user with the given username and password.

Requires an valid access token for the admin user.

Returns a 409 status code if the user already exists.

Parameters:

form_data (Annotated[OAuth2PasswordRequestForm])
session (Annotated[Session])
authenticated_user (Annotated[User])

Return type:

CreateUserResponse

async ogdc_runner.service.auth_routes.get_current_user(current_user: Annotated[User, fastapi.Depends]) → dict[str, str]¶

Return the current authenticated user.

Useful for testing that authentication is working as expected.

Parameters:: current_user (Annotated[User])
Return type:: dict[str, str]

ogdc_runner.service.auth_routes.get_output(recipe_workflow_name: str) → OutputResponse¶

Get a presigned s3 url for the outputs of the given recipe workflow.

TODO: support other output types. This assumes a temporary output stored on Argo’s artifact s3 storage.

Parameters:: recipe_workflow_name (str)
Return type:: OutputResponse

ogdc_runner.service.auth_routes.status(recipe_workflow_name: str) → StatusResponse¶

Check an argo workflow’s status.

Parameters:: recipe_workflow_name (str)
Return type:: StatusResponse

ogdc_runner.service.auth_routes.submit(submit_recipe_request: SubmitRecipeRequest) → SubmitRecipeResponse¶

Submit a recipe to OGDC for execution.

Requires a valid auth token.

Parameters:: submit_recipe_request (SubmitRecipeRequest)
Return type:: SubmitRecipeResponse