How To Create User Defined SQLite Aggregate Function In Python

This article will tell you how to create user-defined SQLite3 aggregation functions in python, it will also show you an example of how to use it.

1. How To Create User-Defined SQLite3 Aggregate Function In Python Steps.

1. Create a python class in a python module ( .py file ).
2. The python class must implement the two functions def step(self, value) and def finalize(self).
3. The step function will be invoked every time for each row of the SQL statement results.
4. The finalize function will invoke once when the aggregation operation complete and return the result.
5. In my example, there are two python classes, one is class MinLen and the other is class MaxLen.
6. They are all defined in the python module /com/dev2qa/example/sqlite/util.py file.

2. How To Use User-Defined SQLite3 Aggregate Function In Python Steps.

1. Open SQLite3 database connection object.
2. Invoke the above connection object’s create_aggregate(aggregation-function-name, parameter-number, python-class) function to register the user-defined aggregation function.

   conn.create_aggregate('min_len', 1, com.dev2qa.example.sqlite.util.MinLen)
       
   conn.create_aggregate('max_len', 1, com.dev2qa.example.sqlite.util.MaxLen)

3. Now you can use the user-defined aggregation function in SQL statement.

3. How To Create & Use SQLite3 User Defined Aggregation Function In Python Examples.

1. There are two python files in this example, /com/dev2qa/example/sqlite/util.py and /com/dev2qa/example/sqlite/SQLiteOperationExample.py.
2. Below is the util.py file source code.

   class MinLen(object):
   
       def __init__(self):
           
           self.min_len = None
           
       def step(self, value):
           
           if self.min_len is None:
               
               self.min_len = value    
               
               return 
           
           if len(self.min_len) >  len(value):
               
               self.min_len = value
               
       def finalize(self):
           
           return self.min_len        
       
       
   class MaxLen(object):
   
       def __init__(self):
           
           self.max_len = None
           
       def step(self, value):
           
           if self.max_len is None:
               
               self.max_len = value    
               
               return 
           
           if len(self.max_len) <  len(value):
               
               self.max_len = value
               
       def finalize(self):
           
           return self.max_len

3. Below is the SQLiteOperationExample.py file source code.

   import sqlite3
   
   import com.dev2qa.example.sqlite.util
   
   db_name = 'test-sqlite.db'
   
   table_name = 'user_account'
   
   def invoke_aggregate_function():
       
       conn = sqlite3.connect(db_name)
       
       conn.create_aggregate('min_len', 1, com.dev2qa.example.sqlite.util.MinLen)
       
       conn.create_aggregate('max_len', 1, com.dev2qa.example.sqlite.util.MaxLen)
       
       cursor = conn.cursor()
       
       sql_select_min_len_passwd = 'select min_len(passwd) from ' + table_name
       
       cursor.execute(sql_select_min_len_passwd)
       
       min_len_passwd = cursor.fetchone()[0]
       
       print('Min length password : ', min_len_passwd)
       
       
       sql_select_max_len_passwd = 'select max_len(passwd) from ' + table_name
       
       cursor.execute(sql_select_max_len_passwd)
       
       max_len_passwd = cursor.fetchone()[0]
       
       print('Max length password : ', max_len_passwd)
       
       conn.commit()
       
       cursor.close()
       
       conn.close()
   
   if __name__ == '__main__':
       
       invoke_aggregate_function()

4. Below is the example execution output.

   Min length password :  97531
   Max length password :  1000098