Hide navigation sidebar
Hide table of contents sidebar
Skip to content
Back to top
View this page
Toggle table of contents sidebar

# pyrate_limiter.buckets.sqlite_bucket module

Bucket implementation using SQLite

class pyrate_limiter.buckets.sqlite_bucket.Queries

  Bases: `object`

  COUNT_ALL = "SELECT COUNT(*) FROM '{table}'"
  COUNT_BEFORE_INSERT = "\n SELECT :interval{index} as interval, COUNT(*) FROM '{table}'\n WHERE item_timestamp >= :current_timestamp - :interval{index}\n "
  COUNT_BEFORE_LEAK = "SELECT COUNT(*) FROM '{table}' WHERE item_timestamp < {current_timestamp} - {interval}"
  CREATE_BUCKET_TABLE = "\n CREATE TABLE IF NOT EXISTS '{table}' (\n name VARCHAR,\n item_timestamp INTEGER\n )\n "
  CREATE_INDEX_ON_TIMESTAMP = "\n CREATE INDEX IF NOT EXISTS '{index_name}' ON '{table_name}' (item_timestamp)\n "
  DROP_INDEX = "DROP INDEX IF EXISTS '{index}'"
  DROP_TABLE = "DROP TABLE IF EXISTS '{table}'"
  FLUSH = "DELETE FROM '{table}'"
  GET_ALL_ITEM = "SELECT * FROM '{table}' ORDER BY item_timestamp ASC"
  GET_FIRST_ITEM = "SELECT name, item_timestamp FROM '{table}' ORDER BY item_timestamp ASC"
  GET_LAG = "\n SELECT (strftime ('%s', 'now') || substr(strftime ('%f', 'now'), 4)) - (\n SELECT item_timestamp\n FROM '{table}'\n ORDER BY item_timestamp\n ASC\n LIMIT 1\n )\n "
  LEAK = 'DELETE FROM "{table}" WHERE rowid IN (\n SELECT rowid FROM "{table}" ORDER BY item_timestamp ASC LIMIT {count});'
  PEEK = 'SELECT * FROM "{table}" ORDER BY item_timestamp DESC LIMIT 1 OFFSET {count}'
  PUT_ITEM = "\n INSERT INTO '{table}' (name, item_timestamp) VALUES %s\n "

class pyrate_limiter.buckets.sqlite_bucket.SQLiteBucket(rates, conn, table, lock=None)

  Bases: `AbstractBucket`

  For sqlite bucket, we are using the sql time function as the clock item’s timestamp wont matter here

  conn
  count()

    Count number of items in the bucket

    Return type:

      `int`

  flush()

    Flush the whole bucket - Must remove failing-rate after flushing

    Return type:

      `None`

  full_count_query
  classmethod init_from_file(rates, table='rate_bucket', db_path=None, create_new_table=True, use_file_lock=False)
    Return type:

      `SQLiteBucket`

  leak(current_timestamp=None)

    Leaking/clean up bucket

    Return type:

      `int`

  lock
  peek(index)

    Peek at the rate-item at a specific index in latest-to-earliest order NOTE: The reason we cannot peek from the start of the queue(earliest-to-latest) is we can’t really tell how many outdated items are still in the queue

    Return type:

      `Optional`[`RateItem`]

  put(item)

    Put an item (typically the current time) in the bucket return true if successful, otherwise false

    Return type:

      `bool`

  rates
  table