-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
issue #158: province functions module
- Loading branch information
Showing
9 changed files
with
291 additions
and
160 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| from psycopg import Cursor | ||
| from psycopg.rows import dict_row | ||
| from psycopg.sql import SQL | ||
|
|
||
|
|
||
| def create_province(cursor: Cursor, name: str) -> dict | None: | ||
| """ | ||
| Inserts a new province record into the database. | ||
| Args: | ||
| cursor: Database cursor object. | ||
| name: Name of the province. | ||
| Returns: | ||
| The inserted province record as a dictionary, or None if failed. | ||
| """ | ||
| query = SQL(""" | ||
| INSERT INTO province (name) | ||
| VALUES (%s) | ||
| RETURNING *; | ||
| """) | ||
| with cursor.connection.cursor(row_factory=dict_row) as new_cur: | ||
| new_cur.execute(query, (name,)) | ||
| return new_cur.fetchone() | ||
|
|
||
|
|
||
| def read_province(cursor: Cursor, province_id: int) -> dict | None: | ||
| """ | ||
| Retrieves a province record by ID. | ||
| Args: | ||
| cursor: Database cursor object. | ||
| province_id: ID of the province. | ||
| Returns: | ||
| The province record as a dictionary, or None if not found. | ||
| """ | ||
| query = SQL("SELECT * FROM province WHERE id = %s;") | ||
| with cursor.connection.cursor(row_factory=dict_row) as new_cur: | ||
| new_cur.execute(query, (province_id,)) | ||
| return new_cur.fetchone() | ||
|
|
||
|
|
||
| def read_all_provinces(cursor: Cursor) -> list[dict]: | ||
| """ | ||
| Retrieves all province records from the database. | ||
| Args: | ||
| cursor: Database cursor object. | ||
| Returns: | ||
| A list of all province records as dictionaries. | ||
| """ | ||
| query = SQL("SELECT * FROM province;") | ||
| with cursor.connection.cursor(row_factory=dict_row) as new_cur: | ||
| new_cur.execute(query) | ||
| return new_cur.fetchall() | ||
|
|
||
|
|
||
| def update_province(cursor: Cursor, province_id: int, name: str) -> dict | None: | ||
| """ | ||
| Updates an existing province record by ID. | ||
| Args: | ||
| cursor: Database cursor object. | ||
| province_id: ID of the province. | ||
| name: New name of the province. | ||
| Returns: | ||
| The updated province record as a dictionary, or None if not found. | ||
| """ | ||
| query = SQL(""" | ||
| UPDATE province | ||
| SET name = %s | ||
| WHERE id = %s | ||
| RETURNING *; | ||
| """) | ||
| with cursor.connection.cursor(row_factory=dict_row) as new_cur: | ||
| new_cur.execute(query, (name, province_id)) | ||
| return new_cur.fetchone() | ||
|
|
||
|
|
||
| def delete_province(cursor: Cursor, province_id: int) -> dict | None: | ||
| """ | ||
| Deletes a province record by ID. | ||
| Args: | ||
| cursor: Database cursor object. | ||
| province_id: ID of the province. | ||
| Returns: | ||
| The deleted province record as a dictionary, or None if not found. | ||
| """ | ||
| query = SQL(""" | ||
| DELETE FROM province | ||
| WHERE id = %s | ||
| RETURNING *; | ||
| """) | ||
| with cursor.connection.cursor(row_factory=dict_row) as new_cur: | ||
| new_cur.execute(query, (province_id,)) | ||
| return new_cur.fetchone() | ||
|
|
||
|
|
||
| def query_provinces(cursor: Cursor, name: str | None = None) -> list[dict]: | ||
| """ | ||
| Queries provinces based on optional filter criteria. | ||
| Args: | ||
| cursor: Database cursor object. | ||
| name: Optional name to filter provinces. | ||
| Returns: | ||
| A list of province records matching the filter criteria as dictionaries. | ||
| """ | ||
| conditions = [] | ||
| parameters = [] | ||
|
|
||
| if name is not None: | ||
| conditions.append("name = %s") | ||
| parameters.append(name) | ||
|
|
||
| where_clause = " WHERE " + " AND ".join(conditions) if conditions else "" | ||
| query = SQL(f"SELECT * FROM province{where_clause};") | ||
|
|
||
| with cursor.connection.cursor(row_factory=dict_row) as new_cur: | ||
| new_cur.execute(query, parameters) | ||
| return new_cur.fetchall() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -6,7 +6,7 @@ | |
| from psycopg import Connection, connect | ||
|
|
||
| from datastore.db.queries.user import register_user | ||
| from fertiscan.db.models import FullLocation, Location, Region | ||
| from fertiscan.db.models import FullLocation, Location, Province, Region | ||
| from fertiscan.db.queries.location import ( | ||
| create_location, | ||
| delete_location, | ||
|
|
@@ -16,11 +16,8 @@ | |
| update_location, | ||
| upsert_location, | ||
| ) | ||
| from fertiscan.db.queries.organization import ( | ||
| new_organization, | ||
| new_organization_info, | ||
| new_province, | ||
| ) | ||
| from fertiscan.db.queries.organization import new_organization, new_organization_info | ||
| from fertiscan.db.queries.province import create_province | ||
| from fertiscan.db.queries.region import create_region | ||
|
|
||
| load_dotenv() | ||
|
|
@@ -45,8 +42,9 @@ def setUp(self): | |
| self.cursor = self.conn.cursor() | ||
|
|
||
| # Create necessary records for testing | ||
| self.province_id = new_province(self.cursor, uuid.uuid4().hex) | ||
| self.region = create_region(self.cursor, "Test Region", self.province_id) | ||
| self.province = create_province(self.cursor, uuid.uuid4().hex) | ||
| self.province = Province.model_validate(self.province) | ||
| self.region = create_region(self.cursor, "Test Region", self.province.id) | ||
| self.region = Region.model_validate(self.region) | ||
|
|
||
| self.inspector_id = register_user(self.cursor, "[email protected]") | ||
|
|
@@ -270,8 +268,9 @@ def test_get_full_location_with_region_and_province(self): | |
| # Create province and region | ||
| province_name = uuid.uuid4().hex | ||
| region_name = "Test Region" | ||
| province_id = new_province(self.cursor, province_name) | ||
| region = create_region(self.cursor, region_name, province_id) | ||
| province = create_province(self.cursor, province_name) | ||
| province = Province.model_validate(province) | ||
| region = create_region(self.cursor, region_name, province.id) | ||
| region = Region.model_validate(region) | ||
|
|
||
| # Create a location with the new region | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.