text: #!/usr/bin/env python
  
  ##
  # Licensed to the Apache Software Foundation (ASF) under one
  # or more contributor license agreements.  See the NOTICE file
  # distributed with this work for additional information
  # regarding copyright ownership.  The ASF licenses this file
  # to you 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 argparse
  import json
  import platform
  import random
  import string
  import tempfile
  import timeit
  import unittest
  import unittest.mock
  
  import avro.datafile
  import avro.io
  import avro.schema
  
  TYPES = ("A", "CNAME")
  SCHEMA = avro.schema.parse(
      json.dumps(
          {
              "type": "record",
              "name": "Query",
              "fields": [
                  {"name": "query", "type": "string"},
                  {"name": "response", "type": "string"},
                  {"name": "type", "type": "string", "default": "A"},
              ],
          }
      )
  )
  READER = avro.io.DatumReader(SCHEMA)
  WRITER = avro.io.DatumWriter(SCHEMA)
  NUMBER_OF_TESTS = 10000
  MAX_WRITE_SECONDS = 3 if platform.python_implementation() == "PyPy" else 1
  MAX_READ_SECONDS = 3 if platform.python_implementation() == "PyPy" else 1
  
  
  try:
      randbytes = random.randbytes  # type: ignore
  except AttributeError:
  
      def randbytes(n):
          """Polyfill for random.randbytes in Python < 3.9"""
          return random.choices(range(256), k=n)
  
  
  class TestBench(unittest.TestCase):
      def test_minimum_speed(self):
          with tempfile.NamedTemporaryFile(suffix="avr") as temp:
              pass
          self.assertLess(
              time_writes(temp.name, NUMBER_OF_TESTS),
              MAX_WRITE_SECONDS,
              f"Took longer than {MAX_WRITE_SECONDS} second(s) to write the test file with {NUMBER_OF_TESTS} values.",
          )
          self.assertLess(
              time_read(temp.name),
              MAX_READ_SECONDS,
              f"Took longer than {MAX_READ_SECONDS} second(s) to read the test file with {NUMBER_OF_TESTS} values.",
          )
  
  
  def rand_name():
      return "".join(random.sample(string.ascii_lowercase, 15))
  
  
  def rand_ip():
      return ".".join(map(str, randbytes(4)))
  
  
  def picks(n):
      return [{"query": rand_name(), "response": rand_ip(), "type": random.choice(TYPES)} for _ in range(n)]
  
  
  def time_writes(path, number):
      with avro.datafile.DataFileWriter(open(path, "wb"), WRITER, SCHEMA) as dw:
          globals_ = {"dw": dw, "picks": picks(number)}
          return timeit.timeit("dw.append(next(p))", number=number, setup="p=iter(picks)", globals=globals_)
  
  
  def time_read(path):
      """
      Time how long it takes to read the file written in the `write` function.
      We only do this once, because the size of the file is defined by the number sent to `write`.
      """
      with avro.datafile.DataFileReader(open(path, "rb"), READER) as dr:
          return timeit.timeit("tuple(dr)", number=1, globals={"dr": dr})
  
  
  def parse_args():  # pragma: no cover
      parser = argparse.ArgumentParser(description="Benchmark writing some random avro.")
      parser.add_argument(
          "--number",
          "-n",
          type=int,
          default=timeit.default_number,
          help="how many times to run",
      )
      return parser.parse_args()
  
  
  def main():  # pragma: no cover
      args = parse_args()
      with tempfile.NamedTemporaryFile(suffix=".avr") as temp:
          pass
      print(f"Using file {temp.name}")
      print(f"Writing: {time_writes(temp.name, args.number)}")
      print(f"Reading: {time_read(temp.name)}")
  
  
  if __name__ == "__main__":  # pragma: no cover
      main()
  