---
# System prepended metadata

title: Akil Ram

---

# Akil Ram
# Questions

1) brain storming
    1) We will not use airflow anymore, am I right?
    2) all task will be run in kubernetes pod
    3) Decide the flow (how flow will work), give an example (bigquery)




Bigquery
[bigquery, sheet]
sheet <-> bigquery (copy data)
service communication happen here

Points:
1) POD: job deployment type (service communication)
2) pod will be in sleep mode


Architechture:
1) decentralize

pubsub: 
 topics and subscribers
 
 server_api - > single_topic -> [bigquery_subscriber, sheet_subs]
 
 metadata  will decide that where to process
 
 
### command
```
docker build -f DockerfileCommon -t startkiller .
```
### Current Scope:
    Task:
        1. Bigquery
            1.1 bigquery_function
            1.2 bigquery_run
            1.3 bigquery_values
            1.4 bigquery_query
            1.5 bigquery_bucket
            
            
    bigquery_query:
     """Execute a query and write results to table.
     
     
[update: 5 july 2022]
Questions:
1. how other service will talk to this job.
2. how pod will trigger.
3. how parameter we will pass.

Answers from the team.
1.Need to research on how to initiliaze a pod will it be command line or through commands 
2.Currently it is command line its in compute engine 
3.Need to research how can we initiliaze with GKE using commands.
4.There will be a micro service that would be running and that will hit the command or whatever task it has to run and how to pass the commandline arguments    with it, that has to be researched.
```
python main.py --project test --dataset test --table test -s '{ "setup":{ "auth":{ "service":"[JSON OR PATH]" }}}'
```

[update: 7 july 2022]

How to pass the service account and config file  
You will get reference to that secret manager entity of it's tuple and then we will decide a secret manager code for getting the value of service account and then passing it to the task itself. 
The Micro service that we will have on the tenant side would need to do that work, so task would have already everything like the JSON file, with that info about the task itself it will aslo set the service account file and then it should come from it.

Need to research on how to initiliaze a pod will it be command line or through commands.
There would be a command to run the pods itself. We Can use both command line and commands. If its a service API ideally we will use GKE API.
Currently it is command line its in compute engine 
Need to research how can we initiliaze with GKE using commands.
There will be a micro service that would be running and that will hit the command or whatever task it has to run and how to pass the commandline arguments with it, that has to be researched.



update 12/07/22

API -> create Batch job pod -> complete task -> release resources

How parameter will be passed?
Ans: 

Deploy pod on different projects?

Can we use cloud Run?
1. we can deploy docker build there
2. pass arguments to that docker
3. it will spin up pod, process and shutdown when no use



How It will run?
steps:
1. get requerst to API
2. API will create pod and submit job to that pod with dynamic arguments
3. when job will complete after some seconds (can set accordingly) will release resources.

```
# save this a simple-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: batch-job-example-1
  labels:
    jobgroup: bigquery-task
spec:
  ttlSecondsAfterFinished: 30
  template:
    metadata:
      name: bigquery-task
      labels:
        jobgroup: jobexample
    spec:
      restartPolicy: OnFailure
      containers:
      - name: my-job
        image: alpine
        imagePullPolicy: IfNotPresent
        command: ['sh', '-c', 'echo First Batch Job; sleep 10']

```

TO create pod
```
kubectl apply -f simple-job.yaml
```



update [14 july 2022]

entry.py
```python
import uuid
import argparse

from task_executer import Kubernetes
from kubernetes import client


if __name__ == "__main__":

    parser = argparse.ArgumentParser("Task Executor")
    parser.add_argument("input_string", help="Input String", type=str)
    args = parser.parse_args()

    job_id = uuid.uuid4()
    pod_id = job_id

    """ Steps 1 to 3 is the equivalent of the ./manifestfiles/shuffler_job.yaml """

    # Kubernetes instance
    k8s = Kubernetes()

    # STEP1: CREATE A CONTAINER
    _image = "shuffler:latest"
    _name = "shuffler"
    _pull_policy = "IfNotPresent"

    shuffler_container = k8s.create_container(_image, _name, _pull_policy, args.input_string)

    # STEP2: CREATE A POD TEMPLATE SPEC
    _pod_name = f"bigquery-{pod_id}"
    _pod_spec = k8s.create_pod_template(_pod_name, shuffler_container)

    # STEP3: CREATE A JOB
    _job_name = f"my-job-{job_id}"
    _job = k8s.create_job(_job_name, _pod_spec)

    # STEP4: CREATE NAMESPACE
    _namespace = "testns"
    k8s.create_namespace(_namespace)

    # STEP5: EXECUTE THE JOB
    batch_api = client.BatchV1Api()
    batch_api.create_namespaced_job(_namespace, _job)

```


shuffler.py
```python=
import random
import logging
import argparse

logging.basicConfig(level=logging.INFO)


def name_shuffler(input_string):

    if isinstance(input_string, str):

        if len(input_string) < 2:
            raise Exception("Must be at least 2 characters.")

        logging.info(f"Original string: {input_string}")
        str_var = list(input_string)
        random.shuffle(str_var)

        logging.info(f"Shuffled string: {''.join(str_var)}")
    else:
        raise Exception("Input must be a string.")


if __name__ == "__main__":

    parser = argparse.ArgumentParser("Shuffler")
    parser.add_argument("input_string", help="Input String", type=str)

    args = parser.parse_args()

    name_shuffler(args.input_string)
```


```dockerfile=
FROM continuumio/miniconda3:4.6.14

RUN set -ex; \
    apt-get update \
    && apt-get -y install --no-install-recommends build-essential=* python-dev=* make=* \
    && rm -rf /var/lib/apt/lists/*

ENV PYTHONDONTWRITEBYTECODE=1

RUN pip install Kubernetes
COPY shuffler.py .


CMD ["python", "-u", "shuffler.py"]
```


kubernetes==24.2.0

task_executer.py

```python=
import logging
import argparse
import uuid

from kubernetes import client
from kubernetes import config

logging.basicConfig(level=logging.INFO)
config.load_kube_config()


class Kubernetes:
    def __init__(self):

        # Init Kubernetes
        self.core_api = client.CoreV1Api()
        self.batch_api = client.BatchV1Api()

    def create_namespace(self, namespace):

        namespaces = self.core_api.list_namespace()
        all_namespaces = []
        for ns in namespaces.items:
            all_namespaces.append(ns.metadata.name)

        if namespace in all_namespaces:
            logging.info(f"Namespace {namespace} already exists. Reusing.")
        else:
            namespace_metadata = client.V1ObjectMeta(name=namespace)
            self.core_api.create_namespace(
                client.V1Namespace(metadata=namespace_metadata)
            )
            logging.info(f"Created namespace {namespace}.")

        return namespace

    @staticmethod
    def create_container(image, name, pull_policy, args):

        container = client.V1Container(
            image=image,
            name=name,
            image_pull_policy=pull_policy,
            args=[args],
            command=["python", "-u", "shuffler.py"],
        )

        logging.info(
            f"Created container with name: {container.name}, "
            f"image: {container.image} and args: {container.args}"
        )

        return container

    @staticmethod
    def create_pod_template(pod_name, container):
        pod_template = client.V1PodTemplateSpec(
            spec=client.V1PodSpec(restart_policy="OnFailure", containers=[container]),
            metadata=client.V1ObjectMeta(name=pod_name, labels={"pod_name": pod_name, "jobgroup": "bigquery-task"}),
        )

        return pod_template

    @staticmethod
    def create_job(job_name, pod_template):
        metadata = client.V1ObjectMeta(name=job_name, labels={"job_name": job_name})

        job = client.V1Job(
            api_version="batch/v1",
            kind="Job",
            metadata=metadata,
            spec=client.V1JobSpec(backoff_limit=0, ttl_seconds_after_finished=30, template=pod_template),
        )

        return job

```

```
python entry.py "Kubernetes"
```



update [15 july 2022]

```cmd=
minikube image load "shuffle:0.1"


```

---------------------------

#### CODE

https://minikube.sigs.k8s.io/docs/start/

```python=
resources=client.V1ResourceRequirements(
            requests={"cpu": "100m", "memory": "200Mi"},
            limits={"cpu": "500m", "memory": "500Mi"}
        )
```

gcloud auth activate-service-account --key-file=/home/akhil/Downloads/spidr-coreapp-dev-ec4223012dbd.json


update [26 july 2022]
---Have to run multiple tasks 


gcloud container clusters get-credentials test --zone=northamerica-northeast2

[04 Aug 2022]------------------

```python=
# import textwrap
#
import sys
from bigquery.helper import initialize_project

from pydantic import BaseModel, Field

from project import project


class BQTaskRequest(BaseModel):
    # bq parameter
    dataset: str = Field(default=None, description='Cloud ID of Google Cloud Project.')
    table: str = Field(default=None)

    # excel parameter
    excel_sheet: str = None
    excel_workbook: str = None
    data_schema: str = None

    # project parameter
    recipe: str = Field(default=None, alias='recipe', description='Path to recipe json file to load.')
    project: str = None
    key: str = Field(default=None, description='API Key of Google Cloud Project.')

    user: str = Field(default=None, description='Path to USER credentials json file.')
    service: str = Field(default=None, description='Path to SERVICE credentials json file.')
    client: dict = Field(default=None, description='Path to CLIENT credentials json file.')
    instance: int = Field(default=1, description='Instance number of the task to run ( for tasks with same name ) '
                                                 'starting at 1.')
    verbose: str = Field(default=None, description='Print all the steps as they happen.')
    force: str = Field(default=None, description='Not used but included for compatiblity with another script.')
    trace_print: str = Field(default=None, description='Execution trace written to stdout.')
    trace_file: str = Field(default=None, description='Execution trace written to file.')
    # other
    csv: str = None
    task_name: str = None

    class Config:
        allow_population_by_field_name = True


if __name__ == '__main__':
    args = sys.argv[1:]
    print("-------", args)
    kwarg = {}
    if args:
        for arg in args:
            key, value = arg.split("=")
            kwarg[key] = value

    job_parameters = BQTaskRequest(**kwarg)
    project.from_parameter(request=job_parameters)
    initialize_project()


    # print("------------------", job_parameters.dict())

    # print(kwarg)
    # kwarg = {'dataset': 'test', 'table': 'test' , "json": '/Users/ramnivasdhaker/work/bq-task/sheet.json'}


```

pydantic==1.9.1

project file
```python!
###########################################################################
#
#  Copyright 2020 Google LLC
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      https://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#
###########################################################################

"""The core singleton class of StarThinker that translates json to python.

Project loads JSON and parameters and combines them for execturion.  It handles
three important concepts:

  1. Load the JSON and make all task parameters available to python scripts.
  2. Load authentication, all three parameters are optional if scripts do not
     use them.  The following parameters can be passed for authentication.

    user.json - user credentials json ( generated from client ), is refreshed
                by StarThinker as required.  Can be provided as a local path
                or a Cloud Bucket Storage path for distributed jobs.

    service.json - service credentials json ( generated from cloud project ).
                   Passed as a local path or an embedded json object for
                   distributed jobs.

    client.json - client credentials json ( generated from cloud project ).
                  Also require a user json path which will be written to after
                  client authnetication.  Once authenticated this client is not
                  required.

    Credentials can be specified in one of three ways for maximum flexibility:

    A. Specify credentials on command line (highest priority if given)
       --user / -u = user credentials path
       --client / -c = client credentials path (requires user credentials path)
       --service / -s = service credentials path

    B. Define credentials paths in JSON (overruled by command line)
       In each json file create the following entry (client, or user, or
       service)

         ```
         {
           "setup":{
             "id":"[cloud project id]",
             "auth":{
               "client":"[/home/.credentials/hello-world_client.json]",
               "service":"[/home/.credentials/hello-world_service.json]",
               "user":"[/home/.credentials/hello-world_user.json]"
             }
          },
         }
         ```

    C. Use default credentials ( lowest priority, last resort )
       If neither the json not the command line provides a path, the
       environmental variable GOOGLE_APPLICATION_CREDENTIALS will be
       used for service accounts.  It is created by google cloud
       utilities.
"""

import os
import re
import json
import pytz
import argparse
import textwrap

from datetime import datetime
from importlib import import_module

from util.debug import starthinker_trace_start


EXIT_ERROR = 1
EXIT_SUCCESS = 0
RE_UUID = re.compile(r'(\s*)("setup"\s*:\s*{)')


def get_project(filepath=None, stringcontent=None):
  """Loads json for Project Class.  Available as helper.

    Able to load JSON with newlines ( strips all newlines before load ).

    Args:
     - filepath: (string) The local file path to the recipe json file to load.

    Returns:
     Json of recipe file.
  """

  if filepath:
    with open(filepath) as recipe_file:
      stringcontent = recipe_file.read()

  try:
    # allow the json to contain newlines for formatting queries and such
    return json.loads(stringcontent.replace('\n', ' '))
  except ValueError as e:
    pos = 0
    for count, line in enumerate(stringcontent.splitlines(), 1):
      # do not add newlines, e.pos was run on a version where newlines were removed
      pos += len(line)
      if pos >= e.pos:
        e.lineno = count
        e.pos = pos
        e.args = (
          'JSON ERROR: %s LINE: %s CHARACTER: %s ERROR: %s LINE: %s' %
          (filepath, count, pos - e.pos, str(e.msg), line.strip()),
        )
        raise


def utc_to_timezone(timestamp, timezone):
  if timestamp:
    return timestamp.replace(tzinfo=pytz.utc).astimezone(
        pytz.timezone(timezone))
  else:
    return None


def is_scheduled(recipe, task={}):
  """Wrapper for day_hour_scheduled that returns True if current time zone safe hour is in recipe schedule.

     Used as a helper for any cron job running projects.  Keeping this logic in
     project
     helps avoid time zone detection issues and scheduling discrepencies between
     machines.

    Args:
      - recipe: (Recipe JSON) The JSON of a recipe.
      - task: ( dictionary / JSON ) The specific task being considered for
        execution.

    Returns:
      - True if task is scheduled to run current hour, else False.
  """

  tz = pytz.timezone(
    recipe.get('setup', {}).get('timezone', 'America/Los_Angeles')
  )
  now_tz = datetime.now(tz)
  day_tz = now_tz.strftime('%a')
  hour_tz = now_tz.hour

  days = recipe.get('setup', {}).get('day', [])
  hours = [
    int(h) for h in task.get(
      'hour', recipe.get('setup', {}).get('hour', [])
    )
  ]

  if days == [] or day_tz in days:
    if hours == [] or hour_tz in hours:
      return True

  return False


class project:
  """A singleton that represents the loaded recipe within python scripts.

  All access to json scripts within StarThinker must pass through the project
  class.  It handles parameters, time zones, permissions management, and
  scheduling overhead. Task function name must match JSON task name.

  Project is meant as the entry point into all StarThinker scripts as follows:

    ```
    from util_test.project import project

    @project.from_parameters
    def task():
      pass # execute work using project.* as data from json

    if __name__ == "__main__":
      task()
    ```

  Project is meant to be used by a helper.

    ```
    import argparse
    from util_test.project import project

    if __name__ == "__main__":

      # custom parameters
      parser = argparse.ArgumentParser()
      parser.add_argument('custom', help='custom parameter to be added to
      standard project set.')

      # initialize project
      project.from_commandline(parser=parser)

      # access arguments
      auth = 'service' if project.args.service else 'user'
      print(project.args.custom)
    ```

  Project can also be initialized directly for non-json tasks:

    ```
    from util_test.project import project

    if __name__ == "__main__":
      var_recipe = '/somepath/recipe.json'
      var_user = '/somepath/user.json'
      var_service = '/somepath/service.json'

      project.initialize(_recipe=var_recipe, _user=var_user,
      _service=var_service, _verbose=True)
    ```

  Attributes:  Dealing with authentication...
      - project: (string) The Google Cloud project id.
      - user: (string) Path to the user credentials json file.  It can also be a
      Google Cloud Bucket path when passed to the class directly.
      - service: (string) Path to the service credentials json file.  It can
      also be a json object when passed to the project class directly.
      - client: (string) Path to the client credentials json file.  It can only
      be a local path.  Dealing with execution data...
      - instance: (integer) When executing all tasks, it is the one based index
      offset of the task to run.
      - date: (date) A specific date or 'TODAY', which is changed to today's
      date, passed to python scripts for reference.
      - hour: (integer) When executing all tasks, it is the hour if spefified
      for a task to execute.  Dealing with debugging...
      - verbose: (boolean) Prints all debug information in StarThinker code.
      See: if project.verbose: print '...'.
      - force: (boolean) For recipes with specific task hours, forces all tasks
      to run regardless of hour specified.
  """

  args = None
  recipe = None
  task = None
  instance = None
  filepath = None
  verbose = False
  force = False
  function = None
  timezone = None
  now = None
  date = None
  hour = None


  @classmethod
  def from_commandline(cls, _task=None, parser=None, arguments=None):
    """Used in StarThinker scripts as entry point for command line calls.

    Loads json for execution.

    Usage example:

    ```
    import argparse
    from util_test.project import project

    if __name__ == "__main__":

      # custom parameters
      parser = argparse.ArgumentParser()
      parser.add_argument('custom', help='custom parameter to be added to
      standard project set.')

      # initialize project
      project.from_commandline(parser=parser)

      # access arguments
      auth = 'service' if project.args.service else 'user'
      print(project.args.custom)
    ```

    Args:
      - parser: (ArgumentParser) optional custom argument parser ( json argument
        becomes optional if not None )
      - arguments: (String) optional list of parameters to use when invoking
        project ( defaults to ALL if set to None )

    Returns:
      Nothing, this manipulates a singleton object.  All calls to project.*
      result in the same object.

    """

    if parser is None:
      parser = argparse.ArgumentParser(
          formatter_class=argparse.RawDescriptionHelpFormatter,
          description=textwrap.dedent("""\
          Command line to execute all tasks in a recipe once. ( Common Entry Point )

          This script dispatches all the tasks in a JSON recipe to handlers in sequence.
          For each task, it calls a subprocess to execute the JSON instructions, waits
          for the process to complete and dispatches the next task, until all tasks are
          complete or a critical failure ( exception ) is raised.

          If an exception is raised in any task, all following tasks are not executed by design.

          Example: python run.py [path to recipe file] --force
          Caution: This script does NOT check if the last job finished, potentially causing overruns.
          Notes:
            - To avoid running the entire script when debugging a single task, the command line
              can easily replace "all" with the name of any "task" in the json.  For example
              python all/run.py project/sample/say_hello.json

            - Can be easily replaced with the following to run only the "hello" task:
              python task/hello/run.py project/sample/say_hello.json

            - Or specified further to run only the second hello task:
              python task/hello/run.py project/sample/say_hello.json -i 2

      """))
      if arguments is None or '-j' in arguments:
        parser.add_argument('json', help='Path to recipe json file to load.')
    else:
      if arguments is None or '-j' in arguments:
        parser.add_argument(
            '--json', '-j', help='Path to recipe json file to load.')

    if arguments is None or '-p' in arguments:
      parser.add_argument(
          '--project',
          '-p',
          help='Cloud ID of Google Cloud Project.',
          default=None)
    if arguments is None or '-k' in arguments:
      parser.add_argument(
          '--key',
          '-k',
          help='API Key of Google Cloud Project.',
          default=None)
    if arguments is None or '-u' in arguments:
      parser.add_argument(
          '--user',
          '-u',
          help='Path to USER credentials json file.',
          default=None)
    if arguments is None or '-s' in arguments:
      parser.add_argument(
          '--service',
          '-s',
          help='Path to SERVICE credentials json file.',
          default=None)
    if arguments is None or '-c' in arguments:
      parser.add_argument(
          '--client',
          '-c',
          help='Path to CLIENT credentials json file.',
          default=None)

    if arguments is None or '-i' in arguments:
      parser.add_argument(
          '--instance',
          '-i',
          help='Instance number of the task to run ( for tasks with same name ) starting at 1.',
          default=1,
          type=int)

    if arguments is None or '-v' in arguments:
      parser.add_argument(
          '--verbose',
          '-v',
          help='Print all the steps as they happen.',
          action='store_true')
    if arguments is None or '-f' in arguments:
      parser.add_argument(
          '--force',
          '-force',
          help='Not used but included for compatiblity with another script.',
          action='store_true')

    if arguments is None or '-tp' in arguments:
      parser.add_argument(
          '--trace_print',
          '-tp',
          help='Execution trace written to stdout.',
          action='store_true')
    if arguments is None or '-tf' in arguments:
      parser.add_argument(
          '--trace_file',
          '-tf',
          help='Execution trace written to file.',
          action='store_true')

    cls.args, unknown = parser.parse_known_args()

    # initialize the project singleton with passed in parameters
    cls.initialize(
        get_project(cls.args.json) if hasattr(cls.args, 'json') else {},
        _task,
        getattr(cls.args, 'instance', None),
        getattr(cls.args, 'project', None),
        getattr(cls.args, 'user', None),
        getattr(cls.args, 'service', None),
        getattr(cls.args, 'client', None),
        getattr(cls.args, 'json', None),
        getattr(cls.args, 'key', None),
        getattr(cls.args, 'verbose', None),
        getattr(cls.args, 'force', None),
        getattr(cls.args, 'trace_print', None),
        getattr(cls.args, 'trace_file', None)
    )

  @classmethod
  def from_parameter(cls, request):
      cls.args = request
      parameters = request.dict()
      print(parameters.get('recipe'))
      cls.initialize(
          get_project(parameters.get('recipe')) if parameters.get('recipe') else {},
          parameters.get('task_name', None),
          parameters.get('instance', None),
          parameters.get('project', None),
          parameters.get('user', None),
          parameters.get('service', None),
          parameters.get('client', None),
          parameters.get('recipe', None),
          parameters.get('key', None),
          parameters.get('verbose', None),
          parameters.get('force', None),
          parameters.get('trace_print', None),
          parameters.get('trace_file', None)
      )

  @classmethod
  def get_task_index(cls):
    i = 0
    for c, t in enumerate(cls.recipe.get('tasks', [])):
      if next(iter(t.keys())) == cls.function:
        i += 1
        if i == cls.instance:
          return c
    return None


  @classmethod
  def get_task(cls):
    #if cls.task is None:
    i = cls.get_task_index()
    cls.task = None if i is None else next(
        iter(cls.recipe['tasks'][i].values()))
    return cls.task


  @classmethod
  def set_task(cls, function, parameters):
    if cls.task is None:
      cls.recipe['tasks'].append({function: parameters})
      cls.function = function
      cls.task = parameters
      cls.instance = 1
    else:
      i = cls.get_task_index()
      cls.recipe['tasks'][i] = {function: parameters}
      cls.function = function
      cls.task = parameters
      cls.instance = sum([
          1 for c, t in enumerate(cls.recipe['tasks'])
          if t == function and c <= i
      ])

  @staticmethod
  def from_parameters(func):
    """Initializes a project singleton for execution by a task.

    Either loads parameters (recipe, instance) passed to task programatically,
    or if no parameters passed attmepts to load them from the command line.
    Uses decorator pattern, task name is inferred from function ebing decorated.

    Args:
      - recipe: (dict) JSON object representing the project ( setup plus at
        least one task )
      - instance: (integer) numeric offset of task to run if multiple calls to
        thsi task exist
    """

    def from_parameters_wrapper(recipe=None, instance=1):
      if recipe:
        project.initialize(
            _recipe=recipe, _task=func.__name__, _instance=instance)
      else:
        project.from_commandline(func.__name__)
      func()

    return from_parameters_wrapper


  @classmethod
  def initialize(cls,
                 _recipe={},
                 _task=None,
                 _instance=1,
                 _project=None,
                 _user=None,
                 _service=None,
                 _client=None,
                 _filepath=None,
                 _key=None,
                 _verbose=False,
                 _force=False,
                 _trace_print=False,
                 _trace_file=False):
    """Used in StarThinker scripts as programmatic entry point.

    Set up the project singleton for execution of a task, be sure to mimic
    defaults in helper
    this function loads credentials from various source ( command line argument,
    json, default credentials )
    it also sets up time zone aware date and various helper flags such as force
    and verbose.

    Usage example:
    ```
       from util_test.project import project

       if __name__ == "__main__":
         user = 'user.json'
         service = 'service.json'
         recipe = {'setup':..., 'tasks':.... }
         project.initialize(
           _recipe=recipe,
           _user=user,
           _service=service,
           _verbose=True
         )
    ```

    Args:
      - _recipe: (dict) JSON object representing the project ( setup plus at
        least one task )
      - _task: (string) Task name form recipe json task list to execute.
      - _instance: (integer) See module description.
      - _project: (string) See module description.
      - _user: (string) See module description.
      - _service: (string) See module description.
      - _client: (string) See module description.
      - _key: (string) See module description.
      - _verbose: (boolean) See module description.
      - _force: (boolean) See module description.
      - _trace_print: (boolean) True if writing execution trace to stdout.
      - _trace_file: (boolean) True if writing execution trace to file ( see
        config.py ).

    Returns:
      Nothing, this manipulates a singleton object.  All calls to project.*
      result in the same object.
    """

    starthinker_trace_start(_trace_print, _trace_file)

    cls.recipe = _recipe
    cls.function = _task
    cls.instance = _instance
    cls.force = _force

    # populates the task variable based on function and instance
    cls.get_task()

    cls.verbose = _verbose
    cls.filepath = _filepath

    # add setup to json if not provided and loads command line credentials if given
    if 'setup' not in cls.recipe:
      cls.recipe['setup'] = {}
    if 'auth' not in cls.recipe['setup']:
      cls.recipe['setup']['auth'] = {}
    if _service:
      cls.recipe['setup']['auth']['service'] = _service
    if _client:
      cls.recipe['setup']['auth']['client'] = _client
    # if user explicity specified by command line
    if _user:
      cls.recipe['setup']['auth']['user'] = _user
    # or if user not given, then try default credentials ( disabled security risk to assume on behalf of recipe )
    #elif not cls.recipe['setup']['auth'].get('user'):
    #  cls.recipe['setup']['auth']['user'] = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS', None)

    # if project id given, use it
    if _project:
      cls.recipe['setup']['id'] = _project

    if _key:
      cls.recipe['setup']['key'] = _key

    # TBD: if project id not given, use service credentials
    #elif not cls.recipe['setup'].get('id') and cls.recipe['setup']['auth'].get('service'):
    # TBD: if project id not given, use client credentials
    #elif not cls.recipe['setup'].get('id') and cls.recipe['setup']['auth'].get('client'):

    cls.id = cls.recipe['setup'].get('id')
    cls.key = cls.recipe['setup'].get('key')

    # find date based on timezone
    cls.timezone = pytz.timezone(
      cls.recipe['setup'].get(
        'timezone',
        'America/Los_Angeles'
      )
    )
    cls.now = datetime.now(cls.timezone)
    cls.date = cls.now.date()
    cls.hour = cls.now.hour

    if cls.verbose:
      print('TASK:', _task or 'all')
      print('DATE:', cls.now.date())
      print('HOUR:', cls.now.hour)


  @classmethod
  def execute(cls, _force=False):
    """Run all the tasks in a project in one sequence.

    ```
    from util_test.project import project

    if __name__ == "__main__":
      var_user = '[USER CREDENTIALS JSON STRING OR PATH]'
      var_service = '[SERVICE CREDENTIALS JSON STRING OR PATH]'
      var_recipe = {
        "tasks":[
          { "dataset":{
            "auth":"service",
            "dataset":"Test_Dataset"
          }}
        ]
      }

      project.initialize(
        _recipe=var_recipe,
        _user=var_user,
        _service=var_service,
        _verbose=True
      )
      project.execute()
    ```

    For a full list of tasks see: scripts/*.json

    """

    returncode = EXIT_SUCCESS
    instances = {}
    for task in cls.recipe['tasks']:
      script, task = next(iter(task.items()))

      # count instance per task
      instances.setdefault(script, 0)
      instances[script] += 1

      print('RUNNING TASK:', '%s %d' % (script, instances[script]))

      if _force or cls.force or is_scheduled(cls.recipe, task):
        try:
          python_callable = getattr(
            import_module('starthinker.task.%s.run' % script),
            script
          )
          python_callable(cls.recipe, instances[script])
        except Exception as e:
          print(str(e))
          returncode = EXIT_ERROR
      else:
        print(
          'Schedule Skipping: add --force to ignore schedule or run specific task handler'
        )

    return returncode

```

helper.py
```python
###########################################################################
#
#  Copyright 2020 Google LLC
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      https://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#
###########################################################################

import json

from project import project
from util.bigquery import get_schema
from util.bigquery import table_to_schema
from util.csv import csv_to_rows
from util.csv import excel_to_rows

from util.bigquery import rows_to_table


def initialize_project():
    auth = 'service' if project.args.service else 'user'
    schema = json.loads(project.args.data_schema) if project.args.data_schema else None

    print(project.args)
    if project.args.csv:

        with open(project.args.csv, 'r') as csv_file:
            rows = csv_to_rows(csv_file.read())

            if not schema:
                rows, schema = get_schema(rows)
                print('DETECETED SCHEMA', json.dumps(schema))
                print('Please run again with the above schema provided.')
                exit()

            rows_to_table(
                auth,
                project.id,
                project.args.dataset,
                project.args.table,
                rows,
                schema
            )

    elif project.args.excel_workbook and project.args.excel_sheet:
        with open(project.args.excel_workbook, 'r') as excel_file_2:
            excel_file = project.args.excel_workbook
            rows = excel_to_rows(excel_file, project.args.excel_sheet)

            if not schema:
                rows, schema = get_schema(rows)
                print('DETECETED SCHEMA', json.dumps(schema))
                print('Please run again with the above schema provided.')
                exit()

            rows_to_table(
                auth,
                project.id,
                project.args.dataset,
                project.args.table,
                rows,
                schema
            )

    else:
        # print schema
        print(json.dumps(
            table_to_schema(
                auth,
                project.id,
                project.args.dataset,
                project.args.table
            ),
            indent=2
        ))

```


```python
###########################################################################
#
#  Copyright 2020 Google LLC
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      https://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#
###########################################################################

"""Handler for "bigquery" task in recipes.

One of the oldest tasks in StarThinker, due for a refactor to get_rows and
put_rows. Please test thouroughly when modifying this module.

Includes:
  bigquery_function - generate custom tables or functions
  bigquery_query - execute a query and write results to table ( future get_rows / put_rows )
   table - write query results to a table
   sheet - write query results to a sheet
   view - write query results to a view
  bigquery_run - execute a query without expected return results
  bigquery_storage - read from storage into a table
  bigquery_values - write explicit values to a table ( future get_rows )
"""

from util.project import project
from util.csv import rows_to_type
from util.sheets import sheets_clear
from util.sheets import sheets_write
from util.data import get_rows, put_rows
from util.bigquery import query_to_table, query_to_view, storage_to_table, query_to_rows, json_to_table, rows_to_table, \
    run_query, query_parameters
from util.bigquery.functions import pearson_significance_test
from util.bigquery.us_geography import US_GEOGRAPHY_DATA, US_GEOGRAPHY_SCHEMA


def bigquery_function():
    """Generate custom tables or functions.

  See: scripts/bigquery_function.json
  """

    if project.verbose:
        print('FUNCTION', project.task['function'])

    if project.task['function'] == 'Pearson Significance Test':
        run_query(
            project.task['auth'],
            project.id,
            pearson_significance_test(),
            False,
            project.task['to']['dataset']
        )

    elif project.task['function'] == 'US Geography':
        json_to_table(
            project.task['auth'],
            project.id,
            project.task['to']['dataset'],
            'US_Geography',
            US_GEOGRAPHY_DATA,
            US_GEOGRAPHY_SCHEMA
        )


def bigquery_run():
    """Execute a query without expected return results.

  See: scripts/bigquery_run_query.json
  """

    if project.verbose:
        print('RUN QUERY', project.task['run']['query'])

    run_query(
        project.task['auth'],
        project.id,
        query_parameters(
            project.task['run']['query'],
            project.task['run'].get('parameters')
        ),
        project.task['run'].get('legacy', True)
    )


def bigquery_values():
    """Write explicit values to a table.

  TODO: Replace with get_rows.

  See: scripts/bigquery_run_query.json
  """

    if project.verbose:
        print('VALUES', project.task['from']['values'])

    rows = get_rows(project.task['auth'], project.task['from'])
    rows_to_table(
        project.task['to'].get('auth', project.task['auth']),
        project.id,
        project.task['to']['dataset'],
        project.task['to']['table'],
        rows,
        project.task.get('schema', []),
        0
    )


def bigquery_query():
    """Execute a query and write results to table.

  TODO: Replace with get_rows and put_rows combination.

  See: scripts/bigquery_query.json
       scripts/bigquery_storage.json
       scripts/bigquery_to_sheet.json
       scripts/bigquery_view.json
  """

    if 'table' in project.task['to']:
        if project.verbose:
            print('QUERY TO TABLE', project.task['to']['table'])

        query_to_table(
            project.task['auth'],
            project.id,
            project.task['to']['dataset'],
            project.task['to']['table'],
            query_parameters(
                project.task['from']['query'],
                project.task['from'].get('parameters')
            ),
            disposition=project.task['write_disposition']
            if 'write_disposition' in project.task
            else 'WRITE_TRUNCATE',
            legacy=project.task['from'].get(
                'legacy',
                project.task['from'].get('useLegacySql', True)
            ),  # DEPRECATED: useLegacySql,
            target_project_id=project.task['to'].get('project_id', project.id)
        )

    elif 'sheet' in project.task['to']:
        if project.verbose:
            print('QUERY TO SHEET', project.task['to']['sheet'])

        rows = query_to_rows(
            project.task['auth'],
            project.id,
            project.task['from']['dataset'],
            query_parameters(
                project.task['from']['query'],
                project.task['from'].get('parameters')
            ),
            legacy=project.task['from'].get('legacy', True)
        )

        # makes sure types are correct in sheet
        rows = rows_to_type(rows)

        sheets_clear(
            project.task['to'].get('auth', project.task['auth']),
            project.task['to']['sheet'],
            project.task['to']['tab'],
            project.task['to'].get('range', 'A2')
        )
        sheets_write(
            project.task['to'].get('auth', project.task['auth']),
            project.task['to']['sheet'],
            project.task['to']['tab'],
            project.task['to'].get('range', 'A2'),
            rows
        )

    elif 'sftp' in project.task['to']:
        if project.verbose:
            print('QUERY TO SFTP')

        rows = query_to_rows(
            project.task['auth'],
            project.id,
            project.task['from']['dataset'],
            query_parameters(
                project.task['from']['query'],
                project.task['from'].get('parameters')
            ),
            legacy=project.task['from'].get('use_legacy_sql', True)
        )

        if rows:
            put_rows(project.task['auth'], project.task['to'], rows)

    else:
        if project.verbose:
            print('QUERY TO VIEW', project.task['to']['view'])

        query_to_view(
            project.task['auth'],
            project.id,
            project.task['to']['dataset'],
            project.task['to']['view'],
            query_parameters(
                project.task['from']['query'],
                project.task['from'].get('parameters')
            ),
            project.task['from'].get(
                'legacy',
                project.task['from'].get('useLegacySql', True)
            ),  # DEPRECATED: useLegacySql
            project.task['to'].get('replace', False)
        )


def bigquery_storage():
    """Read from storage into a table.

  See: scripts/bigquery_storage.json
  """

    if project.verbose:
        print('STORAGE TO TABLE', project.task['to']['table'])

    storage_to_table(
        project.task['auth'], project.id, project.task['to']['dataset'],
        project.task['to']['table'],
        project.task['from']['bucket'] + ':' + project.task['from']['path'],
        project.task.get('schema', []), project.task.get('skip_rows', 1),
        project.task.get('structure', 'CSV'),
        project.task.get('disposition', 'WRITE_TRUNCATE')
    )


def bigquery_bucket():
    """Read from storage into a table.

  See: scripts/bigquery_bucket.json
  """

    if project.verbose:
        print('STORAGE TO TABLE', project.task['to']['table'])

    storage_to_table(
        project.task['auth'], project.id, project.task['to']['dataset'],
        project.task['to']['table'],
        project.task['from']['bucket'] + ':' + project.task['from']['path'],
        project.task.get('schema', []), project.task.get('skip_rows', 1),
        project.task.get('structure', 'CSV'),
        project.task.get('disposition', 'WRITE_TRUNCATE')
    )


@project.from_parameters
def bigquery():
    if 'function' in project.task:
        bigquery_function()
    elif 'run' in project.task and 'query' in project.task.get('run', {}):
        bigquery_run()
    elif 'values' in project.task['from']:
        bigquery_values()
    elif 'query' in project.task['from']:
        bigquery_query()
    elif 'bucket' in project.task['from']:
        bigquery_bucket()
    else:
        raise NotImplementedError('The bigquery task has no such handler.')


if __name__ == '__main__':
    bigquery()

```
``

```python main.py
if __name__ == '__main__':
    args = sys.argv[1:]
    print("-------", args)
    kwarg = {}
    if args:
        for arg in args:
            key, value = arg.split("=")
            kwarg[key] = value

    job_parameters = BQTaskRequest(**kwarg)
    project.from_parameter(request=job_parameters)
    # initialize_project()
    bigquery()
```


```python=
class Singleton(type):
    def __init__(self, name, bases, dic):
        self.__single_instance = None
        super().__init__(name, bases, dic)
 
    def __call__(cls, *args, **kwargs):
        if cls.__single_instance:
            return cls.__single_instance
        single_obj = cls.__new__(cls)
        single_obj.__init__(*args, **kwargs)
        cls.__single_instance = single_obj
        return single_obj
```



``` json=
{
  "setup":{
    "license":"Apache License, Version 2.0",
    "copyright":"Copyright 2020 Google LLC"
  },
  "tasks":[
    { "dataset":{
      "auth":"service",
      "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
      "clear":true
    }},
    { "bigquery": {
      "auth":"service",
      "from":{
        "values":[
          ["2018-02-27", "dog", 7, 67],
          ["2018-03-01", "cat", 5, 1.5],
          ["2018-03-02", "bird", 12, 0.44],
          ["2018-03-03", "lizard", 22, 1],
          ["2018-03-04", "dinosaur", 1600, 273.97]
        ]
      },
      "to":{
        "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
        "table":"BQ_Values"
      },
      "schema":[
        { "name":"Date_Column", "type":"DATE" },
        { "name":"String_Column", "type":"STRING" },
        { "name":"Integer_Column", "type":"INTEGER" },
        { "name":"Float_Column", "type":"FLOAT" }
      ]
    }},
    { "test": {
      "auth":"service",
      "bigquery":{
        "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
        "table":"BQ_Values",
        "schema":[
          { "name":"Date_Column", "type":"DATE" },
          { "name":"String_Column", "type":"STRING" },
          { "name":"Integer_Column", "type":"INTEGER" },
          { "name":"Float_Column", "type":"FLOAT" }
        ],
        "values":[
          ["2018-02-27", "dog", 7, 67.0],
          ["2018-03-01", "cat", 5, 1.5],
          ["2018-03-02", "bird", 12, 0.44],
          ["2018-03-03", "lizard", 22, 1.0],
          ["2018-03-04", "dinosaur", 1600, 273.97]
        ]
      }
    }},
    { "sheets":{
      "auth":"user",
      "sheet":"StarThinker Template: Sheets Test",
      "tab":"Sheet_Template",
      "range":"A1:C",
      "header":true,
      "out":{
        "auth": "service",
        "bigquery":{
          "auth": "service",
          "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
          "table":"Sheet_To_BigQuery",
          "schema":[
            { "name":"Animal", "type":"STRING" },
            { "name":"Age", "type":"INTEGER" },
            { "name":"Weight_lbs", "type":"FLOAT" }
          ]
        }
      }
    }},
    { "test": {
      "auth":"service",
      "bigquery":{
        "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
        "table":"Sheet_To_BigQuery",
        "schema":[
          {"name": "Animal", "type": "STRING"},
          {"name": "Age", "type": "INTEGER"},
          {"name": "Weight_lbs", "type": "FLOAT"}
        ],
        "values":[
          ["dog", 7, 67.0],
          ["cat", 5, 1.5],
          ["bird", 12, 0.44],
          ["lizard", 22, 1.0],
          ["dinosaur", 1600, 273.97]
        ]
      }
    }},
    { "test": {
      "auth":"service",
      "bigquery":{
        "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
        "query":"SELECT * FROM {test_dataset}.Sheet_To_BigQuery;",
        "parameters": {
          "test_dataset": {"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}}
        },
        "values":[
          ["dog", 7, 67.0],
          ["cat", 5, 1.5],
          ["bird", 12, 0.44],
          ["lizard", 22, 1.0],
          ["dinosaur", 1600, 273.97]
        ]
      }
    }},
    { "test": {
      "auth":"service",
      "bigquery":{
        "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
        "query":"SELECT * FROM {test_dataset}.Sheet_To_BigQuery WHERE Animal='dog';",
        "parameters": {
          "test_dataset": {"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}}
        },
        "values":[
          ["dog", 7, 67.0]
        ]
      }
    }},
    { "sheets":{
      "auth":"user",
      "sheet":{"field":{ "name":"test_run_id", "kind":"string", "description":"The sheet to use for the test.", "prefix":"StarThinker Test BigQuery ", "default": "Manual"}},
      "tab":"BigQuery_Test",
      "delete":true,
      "template":{}
    }},
    { "bigquery": {
      "auth":"service",
      "from":{
        "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
        "query":"SELECT * FROM {test_dataset}.Sheet_To_BigQuery WHERE Animal IN ('dog', 'cat') ORDER BY Animal;",
        "parameters": {
          "test_dataset": {"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}}
        }
      },
      "to":{
        "auth": "user",
        "sheet":{"field":{ "name":"test_run_id", "kind":"string", "description":"The sheet to use for the test.", "prefix":"StarThinker Test BigQuery ", "default": "Manual"}},
        "tab":"BigQuery_Test",
        "range":"A1:C"
      }
    }},
    { "test": {
      "auth":"user",
      "sheets": {
        "sheet":{"field":{ "name":"test_run_id", "kind":"string", "description":"The sheet to use for the test.", "prefix":"StarThinker Test BigQuery ", "default": "Manual"}},
        "tab":"BigQuery_Test",
        "range":"A1:C",
        "values":[
          ["cat", 5, 1.5],
          ["dog", 7, 67]
        ]
      }
    }},
    { "drive":{
      "auth":"user",
      "delete":{"field":{ "name":"test_run_id", "kind":"string", "description":"The filename to use for the test.", "prefix":"StarThinker Test BigQuery ", "default": "Manual"}}
    }},
    { "dataset":{
      "auth":"service",
      "dataset":{"field":{ "name":"test_run_id", "kind":"string", "description":"The dataset to use for the test.", "prefix":"StarThinker_Test_BigQuery_", "default": "Manual"}},
      "delete":true
    }}
  ]
}

```


#### API

```python=
import json
recipe_encoded = json.dumps(recipe).encode()
```

Bq Task

```python=
import json
json.loads(recipe.decode())

```


recipe={"setup": {"license": "Apache License, Version 2.0", "copyright": "Copyright 2020 Google LLC"}, "tasks": [{"dataset": {"auth": "service", "dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "clear": true}}]}


{'setup': {'license': 'Apache License, Version 2.0', 'copyright': 'Copyright 2020 Google LLC'}, 'tasks': [{'dataset': {'auth': 'service', 'dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}, 'clear': True}}, {'bigquery': {'auth': 'service', 'from': {'values': [['2018-02-27', 'dog', 7, 67], ['2018-03-01', 'cat', 5, 1.5], ['2018-03-02', 'bird', 12, 0.44], ['2018-03-03', 'lizard', 22, 1], ['2018-03-04', 'dinosaur', 1600, 273.97]]}, 'to': {'dataset': 'test', 'table': 'BQ_Values'}, 'schema': [{'name': 'Date_Column', 'type': 'DATE'}, {'name': 'String_Column', 'type': 'STRING'}, {'name': 'Integer_Column', 'type': 'INTEGER'}, {'name': 'Float_Column', 'type': 'FLOAT'}]}}, {'test': {'auth': 'service', 'bigquery': {'dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}, 'table': 'BQ_Values', 'schema': [{'name': 'Date_Column', 'type': 'DATE'}, {'name': 'String_Column', 'type': 'STRING'}, {'name': 'Integer_Column', 'type': 'INTEGER'}, {'name': 'Float_Column', 'type': 'FLOAT'}], 'values': [['2018-02-27', 'dog', 7, 67.0], ['2018-03-01', 'cat', 5, 1.5], ['2018-03-02', 'bird', 12, 0.44], ['2018-03-03', 'lizard', 22, 1.0], ['2018-03-04', 'dinosaur', 1600, 273.97]]}}}, {'sheets': {'auth': 'user', 'sheet': 'StarThinker Template: Sheets Test', 'tab': 'Sheet_Template', 'range': 'A1:C', 'header': True, 'out': {'auth': 'service', 'bigquery': {'auth': 'service', 'dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}, 'table': 'Sheet_To_BigQuery', 'schema': [{'name': 'Animal', 'type': 'STRING'}, {'name': 'Age', 'type': 'INTEGER'}, {'name': 'Weight_lbs', 'type': 'FLOAT'}]}}}}, {'test': {'auth': 'service', 'bigquery': {'dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}, 'table': 'Sheet_To_BigQuery', 'schema': [{'name': 'Animal', 'type': 'STRING'}, {'name': 'Age', 'type': 'INTEGER'}, {'name': 'Weight_lbs', 'type': 'FLOAT'}], 'values': [['dog', 7, 67.0], ['cat', 5, 1.5], ['bird', 12, 0.44], ['lizard', 22, 1.0], ['dinosaur', 1600, 273.97]]}}}, {'test': {'auth': 'service', 'bigquery': {'dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}, 'query': 'SELECT * FROM {test_dataset}.Sheet_To_BigQuery;', 'parameters': {'test_dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}}, 'values': [['dog', 7, 67.0], ['cat', 5, 1.5], ['bird', 12, 0.44], ['lizard', 22, 1.0], ['dinosaur', 1600, 273.97]]}}}, {'test': {'auth': 'service', 'bigquery': {'dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}, 'query': "SELECT * FROM {test_dataset}.Sheet_To_BigQuery WHERE Animal in ('dog',);", 'parameters': {'test_dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}}, 'values': [['dog', 7, 67.0]]}}}, {'sheets': {'auth': 'user', 'sheet': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The sheet to use for the test.', 'prefix': 'StarThinker Test BigQuery ', 'default': 'Manual'}}, 'tab': 'BigQuery_Test', 'delete': True, 'template': {}}}, {'bigquery': {'auth': 'service', 'from': {'dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}, 'query': "SELECT * FROM {test_dataset}.Sheet_To_BigQuery WHERE Animal IN ('dog', 'cat') ORDER BY Animal;", 'parameters': {'test_dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}}}, 'to': {'auth': 'user', 'sheet': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The sheet to use for the test.', 'prefix': 'StarThinker Test BigQuery ', 'default': 'Manual'}}, 'tab': 'BigQuery_Test', 'range': 'A1:C'}}}, {'test': {'auth': 'user', 'sheets': {'sheet': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The sheet to use for the test.', 'prefix': 'StarThinker Test BigQuery ', 'default': 'Manual'}}, 'tab': 'BigQuery_Test', 'range': 'A1:C', 'values': [['cat', 5, 1.5], ['dog', 7, 67]]}}}, {'drive': {'auth': 'user', 'delete': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The filename to use for the test.', 'prefix': 'StarThinker Test BigQuery ', 'default': 'Manual'}}}}, {'dataset': {'auth': 'service', 'dataset': {'field': {'name': 'test_run_id', 'kind': 'string', 'description': 'The dataset to use for the test.', 'prefix': 'StarThinker_Test_BigQuery_', 'default': 'Manual'}}, 'delete': True}}]}





#### ------------------------ 20 aug 2022 ------------------
```string=
{"setup": {"license": "Apache License, Version 2.0", "copyright": "Copyright 2020 Google LLC"}, "tasks": [{"dataset": {"auth": "service", "dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "clear": true}}, {"bigquery": {"auth": "service", "from": {"values": [["2018-02-27", "dog", 7, 67], ["2018-03-01", "cat", 5, 1.5], ["2018-03-02", "bird", 12, 0.44], ["2018-03-03", "lizard", 22, 1], ["2018-03-04", "dinosaur", 1600, 273.97]]}, "to": {"dataset": "test", "table": "BQ_Values"}, "schema": [{"name": "Date_Column", "type": "DATE"}, {"name": "String_Column", "type": "STRING"}, {"name": "Integer_Column", "type": "INTEGER"}, {"name": "Float_Column", "type": "FLOAT"}]}}, {"test": {"auth": "service", "bigquery": {"dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "table": "BQ_Values", "schema": [{"name": "Date_Column", "type": "DATE"}, {"name": "String_Column", "type": "STRING"}, {"name": "Integer_Column", "type": "INTEGER"}, {"name": "Float_Column", "type": "FLOAT"}], "values": [["2018-02-27", "dog", 7, 67], ["2018-03-01", "cat", 5, 1.5], ["2018-03-02", "bird", 12, 0.44], ["2018-03-03", "lizard", 22, 1], ["2018-03-04", "dinosaur", 1600, 273.97]]}}}, {"sheets": {"auth": "user", "sheet": "StarThinker Template: Sheets Test", "tab": "Sheet_Template", "range": "A1:C", "header": true, "out": {"auth": "service", "bigquery": {"auth": "service", "dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "table": "Sheet_To_BigQuery", "schema": [{"name": "Animal", "type": "STRING"}, {"name": "Age", "type": "INTEGER"}, {"name": "Weight_lbs", "type": "FLOAT"}]}}}}, {"test": {"auth": "service", "bigquery": {"dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "table": "Sheet_To_BigQuery", "schema": [{"name": "Animal", "type": "STRING"}, {"name": "Age", "type": "INTEGER"}, {"name": "Weight_lbs", "type": "FLOAT"}], "values": [["dog", 7, 67], ["cat", 5, 1.5], ["bird", 12, 0.44], ["lizard", 22, 1], ["dinosaur", 1600, 273.97]]}}}, {"test": {"auth": "service", "bigquery": {"dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "query": "SELECT * FROM {test_dataset}.Sheet_To_BigQuery;", "parameters": {"test_dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}}, "values": [["dog", 7, 67], ["cat", 5, 1.5], ["bird", 12, 0.44], ["lizard", 22, 1], ["dinosaur", 1600, 273.97]]}}}, {"test": {"auth": "service", "bigquery": {"dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "query": "SELECT * FROM {test_dataset}.Sheet_To_BigQuery WHERE Animal='dog';", "parameters": {"test_dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}}, "values": [["dog", 7, 67]]}}}, {"sheets": {"auth": "user", "sheet": {"field": {"name": "test_run_id", "kind": "string", "description": "The sheet to use for the test.", "prefix": "StarThinker Test BigQuery ", "default": "Manual"}}, "tab": "BigQuery_Test", "delete": true, "template": {}}}, {"bigquery": {"auth": "service", "from": {"dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "query": "SELECT * FROM {test_dataset}.Sheet_To_BigQuery WHERE Animal IN ('dog', 'cat') ORDER BY Animal;", "parameters": {"test_dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}}}, "to": {"auth": "user", "sheet": {"field": {"name": "test_run_id", "kind": "string", "description": "The sheet to use for the test.", "prefix": "StarThinker Test BigQuery ", "default": "Manual"}}, "tab": "BigQuery_Test", "range": "A1:C"}}}, {"test": {"auth": "user", "sheets": {"sheet": {"field": {"name": "test_run_id", "kind": "string", "description": "The sheet to use for the test.", "prefix": "StarThinker Test BigQuery ", "default": "Manual"}}, "tab": "BigQuery_Test", "range": "A1:C", "values": [["cat", 5, 1.5], ["dog", 7, 67]]}}}, {"drive": {"auth": "user", "delete": {"field": {"name": "test_run_id", "kind": "string", "description": "The filename to use for the test.", "prefix": "StarThinker Test BigQuery ", "default": "Manual"}}}}, {"dataset": {"auth": "service", "dataset": {"field": {"name": "test_run_id", "kind": "string", "description": "The dataset to use for the test.", "prefix": "StarThinker_Test_BigQuery_", "default": "Manual"}}, "delete": true}}]}
```


```python
import os
import re
import json
import pytz
import argparse
import textwrap

from datetime import datetime
from importlib import import_module

from util.debug import starthinker_trace_start


EXIT_ERROR = 1
EXIT_SUCCESS = 0
RE_UUID = re.compile(r'(\s*)("setup"\s*:\s*{)')


def get_project(filepath=None, stringcontent=None):
  """Loads json for Project Class.  Available as helper.

    Able to load JSON with newlines ( strips all newlines before load ).

    Args:
     - filepath: (string) The local file path to the recipe json file to load.

    Returns:
     Json of recipe file.
  """

  if filepath:
    with open(filepath) as recipe_file:
      stringcontent = recipe_file.read()

  try:
    # allow the json to contain newlines for formatting queries and such
    return json.loads(stringcontent.replace('\n', ' '))
  except ValueError as e:
    pos = 0
    for count, line in enumerate(stringcontent.splitlines(), 1):
      # do not add newlines, e.pos was run on a version where newlines were removed
      pos += len(line)
      if pos >= e.pos:
        e.lineno = count
        e.pos = pos
        e.args = (
          'JSON ERROR: %s LINE: %s CHARACTER: %s ERROR: %s LINE: %s' %
          (filepath, count, pos - e.pos, str(e.msg), line.strip()),
        )
        raise


def utc_to_timezone(timestamp, timezone):
  if timestamp:
    return timestamp.replace(tzinfo=pytz.utc).astimezone(
        pytz.timezone(timezone))
  else:
    return None


def is_scheduled(recipe, task={}):
  """Wrapper for day_hour_scheduled that returns True if current time zone safe hour is in recipe schedule.

     Used as a helper for any cron job running projects.  Keeping this logic in
     project
     helps avoid time zone detection issues and scheduling discrepencies between
     machines.

    Args:
      - recipe: (Recipe JSON) The JSON of a recipe.
      - task: ( dictionary / JSON ) The specific task being considered for
        execution.

    Returns:
      - True if task is scheduled to run current hour, else False.
  """

  tz = pytz.timezone(
    recipe.get('setup', {}).get('timezone', 'America/Los_Angeles')
  )
  now_tz = datetime.now(tz)
  day_tz = now_tz.strftime('%a')
  hour_tz = now_tz.hour

  days = recipe.get('setup', {}).get('day', [])
  hours = [
    int(h) for h in task.get(
      'hour', recipe.get('setup', {}).get('hour', [])
    )
  ]

  if days == [] or day_tz in days:
    if hours == [] or hour_tz in hours:
      return True

  return False


class project:
  """A singleton that represents the loaded recipe within python scripts.

  All access to json scripts within StarThinker must pass through the project
  class.  It handles parameters, time zones, permissions management, and
  scheduling overhead. Task function name must match JSON task name.

  Project is meant as the entry point into all StarThinker scripts as follows:

    ```
    from util_test.project import project

    @project.from_parameters
    def task():
      pass # execute work using project.* as data from json

    if __name__ == "__main__":
      task()
    ```

  Project is meant to be used by a helper.

    ```
    import argparse
    from util_test.project import project

    if __name__ == "__main__":

      # custom parameters
      parser = argparse.ArgumentParser()
      parser.add_argument('custom', help='custom parameter to be added to
      standard project set.')

      # initialize project
      project.from_commandline(parser=parser)

      # access arguments
      auth = 'service' if project.args.service else 'user'
      print(project.args.custom)
    ```

  Project can also be initialized directly for non-json tasks:

    ```
    from util_test.project import project

    if __name__ == "__main__":
      var_recipe = '/somepath/recipe.json'
      var_user = '/somepath/user.json'
      var_service = '/somepath/service.json'

      project.initialize(_recipe=var_recipe, _user=var_user,
      _service=var_service, _verbose=True)
    ```

  Attributes:  Dealing with authentication...
      - project: (string) The Google Cloud project id.
      - user: (string) Path to the user credentials json file.  It can also be a
      Google Cloud Bucket path when passed to the class directly.
      - service: (string) Path to the service credentials json file.  It can
      also be a json object when passed to the project class directly.
      - client: (string) Path to the client credentials json file.  It can only
      be a local path.  Dealing with execution data...
      - instance: (integer) When executing all tasks, it is the one based index
      offset of the task to run.
      - date: (date) A specific date or 'TODAY', which is changed to today's
      date, passed to python scripts for reference.
      - hour: (integer) When executing all tasks, it is the hour if spefified
      for a task to execute.  Dealing with debugging...
      - verbose: (boolean) Prints all debug information in StarThinker code.
      See: if project.verbose: print '...'.
      - force: (boolean) For recipes with specific task hours, forces all tasks
      to run regardless of hour specified.
  """

  args = None
  recipe = None
  task = None
  instance = None
  filepath = None
  verbose = False
  force = False
  function = None
  timezone = None
  now = None
  date = None
  hour = None


  @classmethod
  def from_commandline(cls, _task=None, parser=None, arguments=None):
    """Used in StarThinker scripts as entry point for command line calls.

    Loads json for execution.

    Usage example:

    ```
    import argparse
    from util_test.project import project

    if __name__ == "__main__":

      # custom parameters
      parser = argparse.ArgumentParser()
      parser.add_argument('custom', help='custom parameter to be added to
      standard project set.')

      # initialize project
      project.from_commandline(parser=parser)

      # access arguments
      auth = 'service' if project.args.service else 'user'
      print(project.args.custom)
    ```

    Args:
      - parser: (ArgumentParser) optional custom argument parser ( json argument
        becomes optional if not None )
      - arguments: (String) optional list of parameters to use when invoking
        project ( defaults to ALL if set to None )

    Returns:
      Nothing, this manipulates a singleton object.  All calls to project.*
      result in the same object.

    """

    if parser is None:
      parser = argparse.ArgumentParser(
          formatter_class=argparse.RawDescriptionHelpFormatter,
          description=textwrap.dedent("""\
          Command line to execute all tasks in a recipe once. ( Common Entry Point )

          This script dispatches all the tasks in a JSON recipe to handlers in sequence.
          For each task, it calls a subprocess to execute the JSON instructions, waits
          for the process to complete and dispatches the next task, until all tasks are
          complete or a critical failure ( exception ) is raised.

          If an exception is raised in any task, all following tasks are not executed by design.

          Example: python run.py [path to recipe file] --force
          Caution: This script does NOT check if the last job finished, potentially causing overruns.
          Notes:
            - To avoid running the entire script when debugging a single task, the command line
              can easily replace "all" with the name of any "task" in the json.  For example
              python all/run.py project/sample/say_hello.json

            - Can be easily replaced with the following to run only the "hello" task:
              python task/hello/run.py project/sample/say_hello.json

            - Or specified further to run only the second hello task:
              python task/hello/run.py project/sample/say_hello.json -i 2

      """))
      if arguments is None or '-j' in arguments:
        parser.add_argument('json', help='Path to recipe json file to load.')
    else:
      if arguments is None or '-j' in arguments:
        parser.add_argument(
            '--json', '-j', help='Path to recipe json file to load.')

    if arguments is None or '-p' in arguments:
      parser.add_argument(
          '--project',
          '-p',
          help='Cloud ID of Google Cloud Project.',
          default=None)
    if arguments is None or '-k' in arguments:
      parser.add_argument(
          '--key',
          '-k',
          help='API Key of Google Cloud Project.',
          default=None)
    if arguments is None or '-u' in arguments:
      parser.add_argument(
          '--user',
          '-u',
          help='Path to USER credentials json file.',
          default=None)
    if arguments is None or '-s' in arguments:
      parser.add_argument(
          '--service',
          '-s',
          help='Path to SERVICE credentials json file.',
          default=None)
    if arguments is None or '-c' in arguments:
      parser.add_argument(
          '--client',
          '-c',
          help='Path to CLIENT credentials json file.',
          default=None)

    if arguments is None or '-i' in arguments:
      parser.add_argument(
          '--instance',
          '-i',
          help='Instance number of the task to run ( for tasks with same name ) starting at 1.',
          default=1,
          type=int)

    if arguments is None or '-v' in arguments:
      parser.add_argument(
          '--verbose',
          '-v',
          help='Print all the steps as they happen.',
          action='store_true')
    if arguments is None or '-f' in arguments:
      parser.add_argument(
          '--force',
          '-force',
          help='Not used but included for compatiblity with another script.',
          action='store_true')

    if arguments is None or '-tp' in arguments:
      parser.add_argument(
          '--trace_print',
          '-tp',
          help='Execution trace written to stdout.',
          action='store_true')
    if arguments is None or '-tf' in arguments:
      parser.add_argument(
          '--trace_file',
          '-tf',
          help='Execution trace written to file.',
          action='store_true')

    cls.args, unknown = parser.parse_known_args()

    # initialize the project singleton with passed in parameters
    cls.initialize(
        get_project(cls.args.json) if hasattr(cls.args, 'json') else {},
        _task,
        getattr(cls.args, 'instance', None),
        getattr(cls.args, 'project', None),
        getattr(cls.args, 'user', None),
        getattr(cls.args, 'service', None),
        getattr(cls.args, 'client', None),
        getattr(cls.args, 'json', None),
        getattr(cls.args, 'key', None),
        getattr(cls.args, 'verbose', None),
        getattr(cls.args, 'force', None),
        getattr(cls.args, 'trace_print', None),
        getattr(cls.args, 'trace_file', None)
    )

  @classmethod
  def from_parameter(cls, **parameters):
      if 'recipe' in parameters:
          parameters['recipe'] = json.loads(parameters['recipe'])
      else:
          raise ValueError("recipe not found")
      cls.initialize(
          parameters.get('recipe') if parameters.get('recipe') else {},
          parameters.get('task_name', None),
          parameters.get('instance', None),
          parameters.get('project', None),
          parameters.get('user', None),
          parameters.get('service', None),
          parameters.get('client', None),
          parameters.get('key', None),
          parameters.get('verbose', None),
          parameters.get('force', None),
          parameters.get('trace_print', None),
          parameters.get('trace_file', None)
      )

  @classmethod
  def get_task_index(cls):
    i = 0
    for c, t in enumerate(cls.recipe.get('tasks', [])):
      if next(iter(t.keys())) == cls.function:
        i += 1
        if i == cls.instance:
          return c
    return None


  @classmethod
  def get_task(cls):
    #if cls.task is None:
    i = cls.get_task_index()
    cls.task = None if i is None else next(
        iter(cls.recipe['tasks'][i].values()))
    return cls.task


  @classmethod
  def set_task(cls, function, parameters):
    if cls.task is None:
      cls.recipe['tasks'].append({function: parameters})
      cls.function = function
      cls.task = parameters
      cls.instance = 1
    else:
      i = cls.get_task_index()
      cls.recipe['tasks'][i] = {function: parameters}
      cls.function = function
      cls.task = parameters
      cls.instance = sum([
          1 for c, t in enumerate(cls.recipe['tasks'])
          if t == function and c <= i
      ])

  @staticmethod
  def from_parameters(func):
    """Initializes a project singleton for execution by a task.

    Either loads parameters (recipe, instance) passed to task programatically,
    or if no parameters passed attmepts to load them from the command line.
    Uses decorator pattern, task name is inferred from function ebing decorated.

    Args:
      - recipe: (dict) JSON object representing the project ( setup plus at
        least one task )
      - instance: (integer) numeric offset of task to run if multiple calls to
        thsi task exist
    """

    def from_parameters_wrapper(recipe=None, instance=1):
      if recipe:
        project.initialize(
            _recipe=recipe, _task=func.__name__, _instance=instance)
      else:
        project.from_commandline(func.__name__)
      func()

    return from_parameters_wrapper


  @classmethod
  def initialize(cls,
                 _recipe={},
                 _task=None,
                 _instance=1,
                 _project=None,
                 _user=None,
                 _service=None,
                 _client=None,
                 _key=None,
                 _verbose=False,
                 _force=False,
                 _trace_print=False,
                 _trace_file=False):
    """Used in StarThinker scripts as programmatic entry point.

    Set up the project singleton for execution of a task, be sure to mimic
    defaults in helper
    this function loads credentials from various source ( command line argument,
    json, default credentials )
    it also sets up time zone aware date and various helper flags such as force
    and verbose.

    Usage example:
    ```
       from util_test.project import project

       if __name__ == "__main__":
         user = 'user.json'
         service = 'service.json'
         recipe = {'setup':..., 'tasks':.... }
         project.initialize(
           _recipe=recipe,
           _user=user,
           _service=service,
           _verbose=True
         )
    ```

    Args:
      - _recipe: (dict) JSON object representing the project ( setup plus at
        least one task )
      - _task: (string) Task name form recipe json task list to execute.
      - _instance: (integer) See module description.
      - _project: (string) See module description.
      - _user: (string) See module description.
      - _service: (string) See module description.
      - _client: (string) See module description.
      - _key: (string) See module description.
      - _verbose: (boolean) See module description.
      - _force: (boolean) See module description.
      - _trace_print: (boolean) True if writing execution trace to stdout.
      - _trace_file: (boolean) True if writing execution trace to file ( see
        config.py ).

    Returns:
      Nothing, this manipulates a singleton object.  All calls to project.*
      result in the same object.
    """

    starthinker_trace_start(_trace_print, _trace_file)

    cls.recipe = _recipe
    cls.function = _task
    cls.instance = _instance
    cls.force = _force

    # populates the task variable based on function and instance
    cls.get_task()

    cls.verbose = _verbose

    # add setup to json if not provided and loads command line credentials if given
    if 'setup' not in cls.recipe:
      cls.recipe['setup'] = {}
    if 'auth' not in cls.recipe['setup']:
      cls.recipe['setup']['auth'] = {}
    if _service:
      cls.recipe['setup']['auth']['service'] = _service
    if _client:
      cls.recipe['setup']['auth']['client'] = _client
    # if user explicity specified by command line
    if _user:
      cls.recipe['setup']['auth']['user'] = _user
    # or if user not given, then try default credentials ( disabled security risk to assume on behalf of recipe )
    #elif not cls.recipe['setup']['auth'].get('user'):
    #  cls.recipe['setup']['auth']['user'] = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS', None)

    # if project id given, use it
    if _project:
      cls.recipe['setup']['id'] = _project

    if _key:
      cls.recipe['setup']['key'] = _key

    # TBD: if project id not given, use service credentials
    #elif not cls.recipe['setup'].get('id') and cls.recipe['setup']['auth'].get('service'):
    # TBD: if project id not given, use client credentials
    #elif not cls.recipe['setup'].get('id') and cls.recipe['setup']['auth'].get('client'):

    cls.id = cls.recipe['setup'].get('id')
    cls.key = cls.recipe['setup'].get('key')

    # find date based on timezone
    cls.timezone = pytz.timezone(
      cls.recipe['setup'].get(
        'timezone',
        'America/Los_Angeles'
      )
    )
    cls.now = datetime.now(cls.timezone)
    cls.date = cls.now.date()
    cls.hour = cls.now.hour

    if cls.verbose:
      print('TASK:', _task or 'all')
      print('DATE:', cls.now.date())
      print('HOUR:', cls.now.hour)


  @classmethod
  def execute(cls, _force=False):
    """Run all the tasks in a project in one sequence.

    ```
    from util_test.project import project

    if __name__ == "__main__":
      var_user = '[USER CREDENTIALS JSON STRING OR PATH]'
      var_service = '[SERVICE CREDENTIALS JSON STRING OR PATH]'
      var_recipe = {
        "tasks":[
          { "dataset":{
            "auth":"service",
            "dataset":"Test_Dataset"
          }}
        ]
      }

      project.initialize(
        _recipe=var_recipe,
        _user=var_user,
        _service=var_service,
        _verbose=True
      )
      project.execute()
    ```

    For a full list of tasks see: scripts/*.json

    """

    returncode = EXIT_SUCCESS
    instances = {}
    for task in cls.recipe['tasks']:
      script, task = next(iter(task.items()))

      # count instance per task
      instances.setdefault(script, 0)
      instances[script] += 1

      print('RUNNING TASK:', '%s %d' % (script, instances[script]))

      if _force or cls.force or is_scheduled(cls.recipe, task):
        try:
          python_callable = getattr(
            import_module('starthinker.task.%s.run' % script),
            script
          )
          python_callable(cls.recipe, instances[script])
        except Exception as e:
          print(str(e))
          returncode = EXIT_ERROR
      else:
        print(
          'Schedule Skipping: add --force to ignore schedule or run specific task handler'
        )

    return returncode
```



```python
    kwarg['task_name'] = 'bigquery'
    kwarg['instance'] = 1
    kwarg['project'] = 'spidr-coreapp-dev'
    kwarg['service'] = '/home/akhil/Desktop/bq-task/key.json'
```

Update [Aug - 25th]

identify Job_creation what falls under, where to write all the code methods
Inputs needed to create a job(User_creds or identifer string.json file,) Account ID, identify.json
After creation of job, how to pick recipe, Pulling recipes from topic. single or multiple
How to loop through one recipe
Have a json that have all the details, daivik will get a paayload, how 
Recipe execution,

Fetching the recipe from topic, in that looping of tasks, ceate a 
Refresh token from colud bucket
payload:
"tasks": recipe.tasks,
"job_status": recipe.job_status,
"job_utm": recipe.job_utm,
"worker_utm": recipe.worker_utm,
"worker_uid": recipe.worker_uid,
"active": recipe.active,
"manual": recipe.manual,
"id": recipe.id,
"ingredients": ingredients
"user_identifier":
"service_json":

Whats done is not wrong but incomplete.

# 4th OCT
```python=
# pylint:disable=logging-fstring-interpolation
import logging
from typing import List
from google.api_core.exceptions import DeadlineExceeded
from google import pubsub_v1

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class PubSubHandler:
    @staticmethod
    def receive_messages(
            project_id: str, subscription_id: str, number_of_messages: int = 1000) -> List[any]:
        """Receives messages from a pull subscription."""

        subscriber = pubsub_v1.SubscriberClient()
        # The `subscription_path` method creates a fully qualified identifier
        # in the form `projects/{project_id}/subscriptions/{subscription_id}`
        subscription_path = subscriber.subscription_path(project_id, subscription_id)
        logger.info(f"subscription path: {subscription_path}")

        try:
            # Initialize request argument(s)
            request = pubsub_v1.PullRequest({"subscription": subscription_path, "max_messages": number_of_messages})
            response = subscriber.pull(request=request)
            received_messages = response.received_messages
        except DeadlineExceeded:
            received_messages = []
            logger.info('No messages caused error')
        return received_messages

    @staticmethod
    def ack_messages(project_id: str, subscription_id: str, message_ids: List[any]):
        subscriber = pubsub_v1.SubscriberClient()
        subscription_path = subscriber.subscription_path(project_id, subscription_id)

        if len(message_ids) > 0:
            subscriber.acknowledge(subscription=subscription_path, ack_ids=message_ids)
            logger.info(f'acknowledged msg_ids: {message_ids}')
            return True
        return False

```

```python=
PubSubHandler.ack_messages(project_id="", subscription_id=""], message_ids=ack_ids)
```
    
```python=
def pre_process_messages(pulled_messages):
    messages_dict = {}
    if len(pulled_messages) < 1:
        logger.info('no messages in queue...')
        return messages_dict

    for message in pulled_messages:
        raw_data = message.message.data

        decoded_data = raw_data.decode('utf8')
        messages_dict[message.ack_id] =decoded_data

    return messages_dict
```