abstract interface with remote database table
Project description
TableCrow
tablecrow
is an abstraction library over a generalized database table. Currently, tablecrow
offers an abstraction for PostGreSQL tables with
simple PostGIS operations.
pip install tablecrow
Data Model
tablecrow
sees a database schema as a mapping of field names to Python types, and a database record / row as a mapping of field names to values:
from datetime import datetime
fields = {'id': int, 'time': datetime, 'length': float, 'name': str}
record = {'id': 1, 'time': datetime(2020, 1, 1), 'length': 4.4, 'name': 'long boi'}
For databases with a spatial extension, you can use Shapely geometries:
from shapely.geometry import Polygon
fields = {'id': int, 'polygon': Polygon}
record = {'id': 1, 'polygon': Polygon([(-77.1, 39.65), (-77.1, 39.725), (-77.4, 39.725), (-77.4, 39.65), (-77.1, 39.65)])}
Usage
create a simple table (single primary key, no geometries)
from datetime import datetime
from tablecrow import PostGresTable
table = PostGresTable(
hostname='localhost:5432',
database='postgres',
name='testing',
fields={'id': int, 'time': datetime, 'length': float, 'name': str},
primary_key='id',
username='postgres',
password='<password>',
)
# add a list of records
table.insert([
{'id': 1, 'time': datetime(2020, 1, 1), 'length': 4.4, 'name': 'long boi'},
{'id': 3, 'time': datetime(2020, 1, 3), 'length': 2, 'name': 'short boi'},
{'id': 2},
{'id': 15, 'time': datetime(2020, 3, 3)},
])
# set, access, or delete a single record using its primary key value
table[4] = {'time': datetime(2020, 1, 4), 'length': 5, 'name': 'long'}
record = table[3]
del table[2]
# list of records in the table
num_records = len(table)
records = table.records
# query the database with a dictionary, or a SQL `WHERE` clause as a string
records = table.records_where({'name': 'short boi'})
records = table.records_where({'name': None})
records = table.records_where({'name': '%long%'})
records = table.records_where("time <= '20200102'::date")
records = table.records_where("length > 2 OR name ILIKE '%short%'")
# delete records with a query
table.delete_where({'name': None})
create a table with multiple primary key fields
from datetime import datetime
from tablecrow import PostGresTable
table = PostGresTable(
hostname='localhost:5432',
database='postgres',
name='testing',
fields={'id': int, 'time': datetime, 'length': float, 'name': str},
primary_key=('id', 'name'),
username='postgres',
password='<password>',
)
# a compound primary key allows more flexibility in ID
table.insert([
{'id': 1, 'time': datetime(2020, 1, 1), 'length': 4.4, 'name': 'long boi'},
{'id': 1, 'time': datetime(2020, 1, 1), 'length': 3, 'name': 'short boi'},
{'id': 3, 'time': datetime(2020, 1, 3), 'length': 2, 'name': 'short boi'},
{'id': 3, 'time': datetime(2020, 1, 3), 'length': 6, 'name': 'long boi'},
{'id': 2, 'name': 'short boi'},
])
# key accessors must include entire primary key
table[4, 'long'] = {'time': datetime(2020, 1, 4), 'length': 5}
record = table[3, 'long boi']
create a table with geometry fields
the database must have a spatial extension (such as PostGIS) installed
from pyproj import CRS
from shapely.geometry import MultiPolygon, Polygon, box
from tablecrow import PostGresTable
table = PostGresTable(
hostname='localhost:5432',
database='postgres',
name='testing',
fields={'id': int, 'polygon': Polygon, 'multipolygon': MultiPolygon},
primary_key='id',
username='postgres',
password='<password>',
crs=CRS.from_epsg(4326),
)
big_box = box(-77.4, 39.65, -77.1, 39.725)
little_box_inside_big_box = box(-77.7, 39.725, -77.4, 39.8)
little_box_touching_big_box = box(-77.1, 39.575, -76.8, 39.65)
disparate_box = box(-77.7, 39.425, -77.4, 39.5)
big_box_in_utm18n = box(268397.8, 4392279.8, 320292.0, 4407509.6)
multi_box = MultiPolygon([little_box_inside_big_box, little_box_touching_big_box])
table.insert([
{'id': 1, 'polygon': little_box_inside_big_box},
{'id': 2, 'polygon': little_box_touching_big_box},
{'id': 3, 'polygon': disparate_box, 'multipolygon': multi_box},
])
# find all records with any geometry intersecting the given geometry
records = table.records_intersecting(big_box)
# find all records with only specific geometry fields intersecting the given geometry
records = table.records_intersecting(big_box, geometry_fields=['polygon'])
# you can also provide geometries in a different CRS
records = table.records_intersecting(
big_box_in_utm18n,
crs=CRS.from_epsg(32618),
geometry_fields=['polygon'],
)
Extending
to write your own custom table interface, extend DatabaseTable
:
from typing import Any, Mapping, Sequence, Union
from tablecrow.table import DatabaseTable
class CustomDatabaseTable(DatabaseTable):
# mapping from Python types to database types
FIELD_TYPES = {
'NoneType': 'NotImplemented',
'bool': 'NotImplemented',
'float': 'NotImplemented',
'int': 'NotImplemented',
'str': 'NotImplemented',
'bytes': 'NotImplemented',
'date': 'NotImplemented',
'time': 'NotImplemented',
'datetime': 'NotImplemented',
'timedelta': 'NotImplemented',
}
def __init__(self, database: str, name: str, fields: {str: type}, primary_key: Union[str, Sequence[str]] = None, hostname: str = None,
username: str = None, password: str = None, users: [str] = None):
super().__init__(database, name, fields, primary_key, hostname, username, password, users)
raise NotImplementedError('implement database connection and table creation here')
@property
def exists(self) -> bool:
raise NotImplementedError('implement database table existence check here')
@property
def schema(self) -> str:
raise NotImplementedError('implement string generation for the database schema here')
@property
def remote_fields(self) -> {str: type}:
raise NotImplementedError('implement accessor for database fields here')
def records_where(self, where: Union[Mapping[str, Any], str, Sequence[str]]) -> [{str: Any}]:
raise NotImplementedError('implement database record query here')
def insert(self, records: [{str: Any}]):
raise NotImplementedError('implement database record insertion here')
def delete_where(self, where: Union[Mapping[str, Any], str, Sequence[str]]):
raise NotImplementedError('implement database record deletion here')
def delete_table(self):
raise NotImplementedError('implement database table deletion here')
Acknowledgements
The original core code and methodology of tablecrow
was developed for the National Bathymetric Source project under
the Office of Coast Survey of the National Oceanic and Atmospheric Administration (NOAA), a part of the United
States Department of Commerce.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
File details
Details for the file tablecrow-1.2.0.tar.gz
.
File metadata
- Download URL: tablecrow-1.2.0.tar.gz
- Upload date:
- Size: 16.3 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/3.2.0 pkginfo/1.6.1 requests/2.25.0 setuptools/51.0.0 requests-toolbelt/0.9.1 tqdm/4.54.1 CPython/3.8.6
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 0811a69ebe53766a270cf943ab96415e448c6949fea7323f28a6c3cc916d97e9 |
|
MD5 | 7ae442949b9b75f5cc78782b3d7765bb |
|
BLAKE2b-256 | d279335e587120a40ddd92168e7742bced2f35f08ecf7061d091c662734f4129 |
File details
Details for the file tablecrow-1.2.0-py3-none-any.whl
.
File metadata
- Download URL: tablecrow-1.2.0-py3-none-any.whl
- Upload date:
- Size: 18.7 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/3.2.0 pkginfo/1.6.1 requests/2.25.0 setuptools/51.0.0 requests-toolbelt/0.9.1 tqdm/4.54.1 CPython/3.8.6
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 9cc29792795fc4b1bd4007aae9758a60ecf965e1a00b68e4b465cf7fd90af9b4 |
|
MD5 | 9ef88293e2a87d2aee4c6694793bc68d |
|
BLAKE2b-256 | ed61f2500cfcf98b4fbbca2fb421b1c845e66ae3c25fd1b3d63830846079bcb7 |