humble-database

This file will become your README and also the index of your documentation.

Install

pip install humble_database

How to use

Database Example

import pandas as pd
import os
from sqlalchemy import text
from urllib.request import urlretrieve

urlretrieve(
    "http://2016.padjo.org/files/data/starterpack/census-acs-1year/acs-1-year-2015.sqlite",
    filename='acs.db'
)
db = Database(drivername='sqlite',database = 'acs.db')

db.query_to_df("select * from sqlite_schema").head(2)

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

db.query_to_df("""select * from states limit 5""")

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}

db.query_to_records(
    "select * from states limit 2",
)[0]

{'year': 2015, 'name': 'Alabama', 'geo_id': '04000US01', 'total_population': 4858979, 'white': 3204076, 'black': 1296681, 'hispanic': 192870, 'asian': 58918, 'american_indian': 19069, 'pacific_islander': 2566, 'other_race': 5590, 'median_age': 38.7, 'total_households': 1846390, 'owner_occupied_homes_median_value': 134100, 'per_capita_income': 44765, 'median_household_income': 44765, 'below_poverty_line': 876016, 'foreign_born_population': 169972, 'state': '01'}

ORM Example

SQL Alchemy Models

from pydantic import BaseModel,computed_field,field_validator,ConfigDict,Field
from sqlalchemy import ForeignKey
from sqlalchemy.orm import DeclarativeBase,Mapped, mapped_column,relationship
from typing import List

class Base(DeclarativeBase):
    year:  Mapped[int]
    name: Mapped[str]
    geo_id: Mapped[str]
    total_population: Mapped[int]
    white: Mapped[int]
    black: Mapped[int]
    hispanic: Mapped[int]
    asian: Mapped[int]
    american_indian: Mapped[int]
    pacific_islander: Mapped[int]
    other_race: Mapped[int]
    median_age: Mapped[int]
    total_households: Mapped[int]
    owner_occupied_homes_median_value: Mapped[int]
    per_capita_income: Mapped[int]
    median_household_income: Mapped[int]
    below_poverty_line: Mapped[int]
    foreign_born_population: Mapped[int]

class State(Base):
    __tablename__ = 'states'
    state: Mapped[str] = mapped_column(primary_key=True)
    total_population: Mapped[int]

    places: Mapped[List['Place']] = relationship(back_populates='state_')
    congressional_districts: Mapped[List['CongressionalDistrict']] = relationship(back_populates='state_')

class Place(Base):
    __tablename__ = 'places'
    place: Mapped[str] = mapped_column(primary_key=True)
    total_population: Mapped[int]
    state: Mapped[str] = mapped_column(ForeignKey("states.state"))
    
    state_: Mapped['State'] = relationship(back_populates='places')

class CongressionalDistrict(Base):
    __tablename__ = 'congressional_districts'
    
    congressional_district: Mapped[str] = mapped_column(primary_key=True)
    state: Mapped[str] = mapped_column(ForeignKey("states.state"))
    
    state_: Mapped['State'] = relationship(back_populates='congressional_districts')

with db.session_scope() as session:
    s = session.query(State).first()
    print(s,'\n')
    for place in s.places:
        print(place.name,'::',place.median_household_income)

<__main__.State object> 

Birmingham city, Alabama :: 32378
Dothan city, Alabama :: 44208
Hoover city, Alabama :: 77365
Huntsville city, Alabama :: 46769
Mobile city, Alabama :: 38678
Montgomery city, Alabama :: 41836
Tuscaloosa city, Alabama :: 44125

with db.session_scope() as session:
    result = session.query(State).limit(7).all()
    for state in result:
        print(
            state.name,
            len(state.places),
            len(state.congressional_districts)
        )

Alabama 7 7
Alaska 1 1
Arizona 16 9
Arkansas 6 4
California 137 53
Colorado 16 7
Connecticut 8 5

Pydantic Models

class ACSBase(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    
    year: int = Field()
    name: str = Field()
    geo_id: str = Field()
    total_population: Optional[int] = Field(None)
    white: Optional[int] = Field(None)
    black: Optional[int] = Field(None)
    hispanic: Optional[int] = Field(None)
    asian: Optional[int] = Field(None)
    american_indian: Optional[int] = Field(None)
    pacific_islander: Optional[int] = Field(None)
    other_race: Optional[int] = Field(None)
    median_age: float = Field()
    total_households: Optional[int] = Field(None)
    owner_occupied_homes_median_value: int = Field()
    per_capita_income: int = Field()
    median_household_income: int = Field()
    below_poverty_line: Optional[int] = Field(None)
    foreign_born_population: Optional[int] = Field(None)    
    state: int = Field()

class PlaceModel(ACSBase):
    """A Model for a record from the 'places' table"""
    place: str

class CDModel(ACSBase):
    """A Model for a record from the 'congressional_districts' table"""
    congressional_district: str 

class StateModel(ACSBase):
    """A Model for a record from the 'states' table"""

    places: List[PlaceModel]
    congressional_districts: List[CDModel]
    
    @computed_field(return_type=float,title='People per District',)
    def avg_people_per_cd(self) -> float:
        return sum([cd.total_population for cd in self.congressional_districts]) / len(self.congressional_districts)

from IPython.display import JSON

# mode = serialization includes computed fields
JSON(StateModel.model_json_schema(mode='serialization'))

<IPython.core.display.JSON object>

from humble_database.data_model import DataModel

ACSDataModel = DataModel[StateModel]
JSON(ACSDataModel.model_json_schema(mode='serialization'))

<IPython.core.display.JSON object>

with db.session_scope() as session:
    orm_result = session.query(State).all()
    result = ACSDataModel(data=orm_result)

result

.dataframe tbody tr th {
    vertical-align: top;
}

.dataframe thead th {
    text-align: right;
}