Skip to content

Instantly share code, notes, and snippets.

@ArthurDelannoyazerty
Last active July 1, 2025 12:56
Show Gist options
  • Save ArthurDelannoyazerty/eb3f43ba08834c15a6fc2ff3abb32f4d to your computer and use it in GitHub Desktop.
Save ArthurDelannoyazerty/eb3f43ba08834c15a6fc2ff3abb32f4d to your computer and use it in GitHub Desktop.
Communicate easily with a sqp database
import psycopg2 # pip install psycopg2-binary
from typing import Optional
class InterfaceSQL:
def __init__(self, database:str, host:str, port:str, user:str, password:str):
conn = psycopg2.connect(
database=database,
host=host,
port=port,
user=user,
password=password
)
self.conn = conn
def send_query(self, query:str, variable:tuple|None=None, results:bool=False) -> Optional[list[tuple]]:
"""Send a command to the SQL database.
:param str query: The query command with `%s` for parameters to fill.
:param tuple variable: The parameters that will takes place in the `%s` of the query. The number of `%s` and parameters must be equal.
:param bool results: True if some values will be returned, False if no value is returned, defaults to `False`
:return list[tuple]: A list of tuple that correspond to the output grid of the request.
## Examples
```python
interface_sql.send_query(
"SELECT * FROM column_name,
results=True
)
```
```python
interface_sql.send_query(
"INSERT INTO column_name (%s, %s) VALUES (%s, %s)",
('column1', 'column2', 'value1', 'value2')
)
```
"""
with self.conn.cursor() as cur:
cur.execute(query, variable)
self.conn.commit()
if results:
results = cur.fetchall()
return results
if __name__ == "__main__":
interface_sql = InterfaceSQL('db_name', 'ip', 'port', 'username', 'password')
query = "SELECT * FROM grid LIMIT 10;"
response = interface_sql.send_query(query, results=True)
print(response)
query = "INSERT INTO grid (face, face_i, face_j, cuboid_i, cuboid_j, geom) VALUES (%s, %s, %s, %s, %s, %s) RETURNING id;"
response = interface_sql.send_query(
query,
('0',
'1',
'2',
'3',
'4',
'POLYGON ((0 0 0,4 0 0,4 4 0,0 4 0,0 0 0),(1 1 0,2 1 0,2 2 0,1 2 0,1 1 0))'),
results=True)
print(response)
query = "SELECT * FROM grid WHERE id=%s;"
response = interface_sql.send_query(query, ('1'), results=True)
print(response)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment