Connect
- class pypsql.connect.DatabaseConnector(path=PosixPath('/home/runner/work/pypsql/pypsql/docs'), db_credential_file='.env')[source]
Create a PostgreSQL connector backed by SQLAlchemy with a queued connection pool.
This class reads connection parameters from a credentials file (see
get_credentials) and constructs an SQLAlchemy engine. It provides helpers to execute queries, push pandas DataFrames, and run SQL scripts.- Parameters:
path (pathlib.Path, optional) – Base path where SQL files and the credentials file live. Defaults to the directory of this module.
db_credential_file (str, optional) – Filename of the credentials file within
path. Defaults to'_credentials.py'.
- path
Base path for SQL files and credentials.
- Type:
pathlib.Path
- db_credential_file
Credentials filename.
- Type:
str
- server
Database host.
- Type:
str
- port
Database port.
- Type:
str
- name_database
Database name.
- Type:
str
- name_user
Username.
- Type:
str
- password_user
Password.
- Type:
str
- engine
Configured SQLAlchemy engine using
QueuePool(pool_size=10).- Type:
sqlalchemy.engine.Engine
Example
dbc = DatabaseConnector() with dbc.start_engine() as conn: rows = conn.execute(text("SELECT 1")).all()
- __init__(path=PosixPath('/home/runner/work/pypsql/pypsql/docs'), db_credential_file='.env')[source]
- drop_table(schema, table)[source]
Drop a table if it exists.
- Parameters:
schema (str) – Schema name.
table (str) – Table name.
- Returns:
None
Example
dbc.drop_table("public", "staging_temp")
- execute_script(sql_script)[source]
Execute an arbitrary SQL script within a transaction.
- Parameters:
sql_script (str) – A full SQL statement or multi-statement script.
- Returns:
None
Example
dbc.execute_script(""" CREATE TABLE IF NOT EXISTS public.example(id int primary key); INSERT INTO public.example VALUES (1) ON CONFLICT DO NOTHING; """)
- get_data(sql_file, replace_dict={}, outcommenting=[])[source]
Fetch data using an external SQL file or a raw SQL string with simple templating.
You can pass either:
the path (relative to
self.path) to a.sqlfile, ora raw SQL string (multiline strings are treated as SQL).
Two templating conventions are supported:
%KEY→ replaced with a quoted value whenreplace_dict['KEY']is a string, or with the plain value for non-strings.§KEY→ replaced with the unquoted value (useful for identifiers or lists).
You can also outcomment lines containing a key and a percent marker by listing the key in
outcommenting. Matching lines are prefixed with--only if that key has not already been replaced.- Parameters:
sql_file (str) – Relative path to a SQL file under
self.pathor a raw SQL string (multiline).replace_dict (dict, optional) – Mapping of placeholder keys to replacement values. Use
%KEYfor auto-quoting strings,§KEYfor raw insertion. Defaults to{}.outcommenting (list, optional) – Keys whose matching lines (containing the key and a
%) should be commented out. Defaults to[].
- Returns:
Query results.
- Return type:
pandas.DataFrame
Examples
Replace a scalar:
df = dbc.get_data("queries/get_users.sql", replace_dict={"user_id": 42})
Insert a raw identifier or list:
df = dbc.get_data("q.sql", replace_dict={"schema": "public"}) # In SQL: SELECT * FROM §schema.users -> SELECT * FROM public.users
Outcomment a filter:
df = dbc.get_data("q.sql", outcommenting=["limit_clause"]) # A line like: AND users.age > %limit_clause -> --AND users.age > %limit_clause
Notes
Replacements are simple regex substitutions; ensure your placeholders do not collide with SQL content.
Prefer parameterized queries for user-supplied data when possible.
Security
This helper performs textual substitution. Avoid injecting untrusted input into identifiers or raw fragments (
§KEY). For dynamic values, prefer%KEYwith safe literals or use SQLAlchemy parameters.
- is_multiline(string)[source]
Return True if the string contains at least one newline.
- Parameters:
string (str) – String to test.
- Returns:
Trueif multiline, elseFalse.- Return type:
bool
- push_data(df, schema, table, if_exists='replace', index=True)[source]
Write a pandas DataFrame to a PostgreSQL table.
- Parameters:
df (pandas.DataFrame) – The DataFrame to persist.
schema (str) – Target schema name.
table (str) – Target table name.
if_exists (str, optional) – One of
'replace','fail', or'append'. Defaults to'replace'.index (bool, optional) – Whether to write the DataFrame index. Defaults to
True.
- Returns:
None
Example
dbc.push_data(df, schema="public", table="events", if_exists="append")
Notes
Uses
engine.begin()for transactional writes and setsindex_label='idx'whenindex=True.
- pypsql.connect.get_credentials(path, filename)[source]
Load database credentials from a simple key=value file, or interactively prompt the user.
The credentials file is expected to contain one key=value pair per line. Values may be single-quoted. Example:
server=localhost port=5432 name_database='mydb' name_user='alice' password_user='s3cr3t'
If the file is missing, the user is prompted for each required field.
- Parameters:
path (pathlib.Path) – Directory containing the credentials file.
filename (str) – Name of the credentials file to read.
- Returns:
- A dictionary with keys:
server(str)port(str)name_database(str)name_user(str)password_user(str)
- Return type:
dict
Notes
Trailing quotes in values are stripped.
Empty or malformed lines are ignored.
Example
creds = get_credentials(Path.cwd(), "_credentials.py") uri = f"postgresql://{creds['name_user']}:{creds['password_user']}@{creds['server']}:{creds['port']}/{creds['name_database']}"