Skip to content

Instantly share code, notes, and snippets.

@d3rpd3rp
Created December 4, 2024 17:14
Show Gist options
  • Save d3rpd3rp/244eb8c0aea66ce12c5696644f2cb772 to your computer and use it in GitHub Desktop.
Save d3rpd3rp/244eb8c0aea66ce12c5696644f2cb772 to your computer and use it in GitHub Desktop.
example class for cursor factory
import atexit
import signal
import sys
from duckdb import connect, BinderException
class MotherDuckClient:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
try:
cls._instance._connection = connect('md:', config = {})
except Exception as e:
print(f'Error connecting to MotherDuck, {e}')
return cls._instance
def get_cursor(self, share_alias: str):
c = self._connection.cursor()
try:
c.execute(f"ATTACH 'md:_share/s2/7564f992-2f93-4799-bce2-1637445b2881' as {share_alias};")
except BinderException as be:
print(f'{be}')
try:
return self._connection.cursor()
except Exception as e:
print(f"Error inst. MotherDuck cursor: {e}")
return c
def close_connection(self):
if self._connection:
try:
self._connection.close()
print("MotherDuck connection closed.")
except Exception as e:
print(f"Error closing MotherDuck connection: {e}")
if __name__ == '__main__':
share_alias = 'share2'
c = MotherDuckClient()
cur = c.get_cursor(share_alias)
cur.execute(f'use {share_alias};')
r = cur.execute(f'from {share_alias}.main.t1 limit 1;').fetchone()
print(r)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment