Source code for q2_sdk.hq.db.auth_token

from argparse import _SubParsersAction
from datetime import datetime
from functools import partial
from typing import List, Optional

from dateutil import parser as datetime_parser
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
    api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)

from .db_object import DbObject
from .representation_row_base import RepresentationRowBase




class AuthTokenRow(RepresentationRowBase):
    # object name: type hinting = "column name in the db response"
    AuthTokenID: IntElement = "AuthTokenID"
    UserID: IntElement = "UserID"
    CreateDate: StringElement = "CreateDate"
    DeletedDate: StringElement = "DeletedDate"
    Serial: StringElement = "Serial"
    FirstUse: StringElement = "FirstUse"





class AuthToken(DbObject):
    # GET_BY_NAME_KEY = "column in the db response"
    NAME = "AuthToken"
    REPRESENTATION_ROW_CLASS = AuthTokenRow



    def add_arguments(self, parser: _SubParsersAction):
        subparser = parser.add_parser("get_auth_token")
        subparser.set_defaults(parser="get_auth_token")
        subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
        subparser.add_argument("user_id", type=int, help="Q2_User.UserID")
        subparser.add_argument("-d", "--include_deleted", action="store_true")

        subparser = parser.add_parser("add_auth_token")
        subparser.set_defaults(parser="create_auth_token")
        subparser.set_defaults(func=partial(self.create, serialize_for_cli=True))
        subparser.add_argument("user_id", type=int, help="Q2_AuthToken.UserID")
        subparser.add_argument("token_serial", type=str, help="Q2_AuthToken.Serial")
        subparser.add_argument(
            "-f", "--first_use", type=datetime_parser.parse, help="Q2_AuthToken.Serial"
        )

        subparser = parser.add_parser("remove_auth_token")
        subparser.set_defaults(parser="delete_auth_token")
        subparser.set_defaults(func=partial(self.delete))
        subparser.add_argument(
            "auth_token_id", type=int, help="Q2_AuthToken.AuthTokenID"
        )




    async def get(
        self, user_id: int, include_deleted=False, serialize_for_cli=False
    ) -> List[AuthTokenRow]:
        assert isinstance(user_id, int), "Please supply a valid user id"

        parameters = [
            ExecuteStoredProcedure.SqlParam(
                ExecuteStoredProcedure.DataType.Int, "user_id", user_id
            )
        ]

        if include_deleted:
            parameters.append(
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Bit,
                    "include_deleted",
                    include_deleted,
                )
            )
        response = await self.call_hq(
            "sdk_GetAuthToken", ExecuteStoredProcedure.SqlParameters(parameters)
        )

        if serialize_for_cli:
            columns = ["AuthTokenID", "UserID", "CreateDate", "Serial", "FirstUse"]
            if include_deleted:
                columns.append("DeletedDate")
            response = self.serialize_for_cli(response, columns)

        return response




    async def create(
        self,
        user_id: int,
        token_serial: str,
        first_use: Optional[datetime] = None,
        serialize_for_cli=True,
    ):
        assert isinstance(user_id, int), "Please supply a valid user id"
        assert isinstance(token_serial, str), "Please supply a valid token serial"

        parameters = [
            ExecuteStoredProcedure.SqlParam(
                ExecuteStoredProcedure.DataType.Int, "user_id", user_id
            ),
            ExecuteStoredProcedure.SqlParam(
                ExecuteStoredProcedure.DataType.VarChar, "serial", token_serial
            ),
        ]

        if first_use:
            assert isinstance(first_use, datetime), (
                "first_use, if provided, must be a datetime instance"
            )
            parameters.append(
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Date,
                    "first_use",
                    first_use.strftime("%Y-%m-%dT%H:%M:%S"),
                )
            )
        response = await self.call_hq(
            "sdk_AddAuthToken", ExecuteStoredProcedure.SqlParameters(parameters)
        )

        if serialize_for_cli:
            columns = ["AuthTokenID", "UserID", "CreateDate", "Serial", "FirstUse"]
            response = self.serialize_for_cli(response, columns)

        return response




    async def delete(self, auth_token_id: int):
        return await self.call_hq(
            "sdk_RemoveAuthToken",
            ExecuteStoredProcedure.SqlParameters([
                ExecuteStoredProcedure.SqlParam(
                    ExecuteStoredProcedure.DataType.Int,
                    "auth_token_id",
                    auth_token_id,
                )
            ]),
        )