Cloud Pub/Sub Publisher Script

publisher.py

import argparse
import dataclasses
import datetime
import json
import random
import time
import uuid
from collections.abc import Generator

from google.cloud import pubsub


@dataclasses.dataclass
class PublishData:
    uid: str
    datetime: str
    no: str
    value: str

    def to_str(self) -> str:
        return json.dumps(dataclasses.asdict(self))


def get_no() -> Generator:
    i = 0
    while True:
        yield i % 3
        i += 1


class GenData:
    no = get_no()

    @classmethod
    def gen_data(cls) -> PublishData:
        now = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))  # noqa: E501
        ret = PublishData(
            uid=str(uuid.uuid4()),
            datetime=now.strftime("%Y-%m-%d %H:%M:%S"),
            no=str(next(cls.no)),
            value=str(random.randint(0, 100))
        )
        return ret


def main(args):
    publish_client = pubsub.PublisherClient()
    topic = f"projects/{args.project}/topics/{args.topic}"

    while True:
        data = GenData.gen_data()
        data_byte = data.to_str().encode("utf-8")
        publish_client.publish(topic, data_byte, org_uid=data.uid)  # noqa: E501
        print(data_byte)
        time.sleep(10)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--project", type=str, required=True)
    parser.add_argument("--topic", type=str, required=True)
    args = parser.parse_args()

    main(args)