androiddrew
/
molten_exp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Initial commit

Browse Source
master
androiddrew 6 years ago
commit f9c9146190
15 changed files with 948 additions and 0 deletions
Show Stats Download Patch File Download Diff File

73
.gitignore vendored
Unescape Escape View File

@ -0,0 +1,73 @@
# virtualenv
env
# Pycharm
.idea
# ---> Python
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
pypyenv/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*,cover
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
# Sphinx documentation
docs/_build/
# Redis
/dump.rdb
# Project
config.py
# Mac
.DS_Store

16
README.md
Unescape Escape View File

@ -0,0 +1,16 @@
# Market Look
A application of tracking in store promotion compliance at a store.
## API
This api returns resource representations with `Content Type` of `application/json`. No other mime types are supported by this api at this time.
### Date times
When a date time is provided with a representation it will follow the ISO 8601 extend format for easy human readability.
### Features
- [ ] Resource expansion.
- [ ] JWT Authentication

5
dev_requirements.in
Unescape Escape View File

@ -0,0 +1,5 @@
black
pip-tools
pytest
pytest-cov
werkzeug

25
dev_requirements.txt
Unescape Escape View File

@ -0,0 +1,25 @@
#
# This file is autogenerated by pip-compile
# To update, run:
#
# pip-compile --output-file dev_requirements.txt dev_requirements.in
#
--trusted-host pypi.python.org
--trusted-host git.aigalatic.com
appdirs==1.4.3 # via black
atomicwrites==1.1.5 # via pytest
attrs==18.1.0 # via black, pytest
black==18.6b4
click==6.7 # via black, pip-tools
coverage==4.5.1 # via pytest-cov
first==2.0.1 # via pip-tools
more-itertools==4.3.0 # via pytest
pip-tools==2.0.2
pluggy==0.7.1 # via pytest
py==1.5.4 # via pytest
pytest-cov==2.5.1
pytest==3.7.1
six==1.11.0 # via more-itertools, pip-tools, pytest
toml==0.9.4 # via black
werkzeug==0.14.1

0
market_look/__init__.py
Unescape Escape View File

230
market_look/app.py
Unescape Escape View File

@ -0,0 +1,230 @@
import os
from typing import List, Tuple, Any, Dict, Optional
from molten import (
annotate,
App,
Request,
QueryParam,
Route,
Settings,
SettingsComponent,
ResponseRendererMiddleware,
HTTP_201,
HTTP_202,
HTTP_404,
HTTPError,
)
from molten.openapi import Metadata, OpenAPIHandler, OpenAPIUIHandler
from molten.contrib.sqlalchemy import (
SQLAlchemyEngineComponent,
SQLAlchemySessionComponent,
SQLAlchemyMiddleware,
)
from .errors import EntityNotFound
from .manager import (
AlertManager,
StoreManager,
GeographyManager,
AlertManagerComponent,
StoreManagerComponent,
GeographyManagerComponent,
)
from .schema import Alert, Store, Geography, APIResponse
from .util import ExtJSONRender
class MarketLook(App):
def handle_404(self, request: Request) -> Tuple[str, APIResponse]:
return (
HTTP_404,
APIResponse(
status=404,
message=f"The resource you are looking for {request.scheme}://{request.host}{request.path} doesn't exist",
),
)
# We are passing explicitly the engine param for
# establishing the utc as the timezone for our connection.
settings = Settings(
{
"database_engine_dsn": os.getenv(
"DATABASE_DSN", "postgres://molten:local@localhost/market_look"
),
"database_engine_params": {
"echo": True,
"connect_args": {"options": "-c timezone=utc"},
},
}
)
get_schema = OpenAPIHandler(
metadata=Metadata(
title="Market Look API",
description="An API for managing promotion compliance at store.",
version="0.1.0",
)
)
get_docs = OpenAPIUIHandler()
setattr(get_docs, "openapi_tags", ["API Management"])
def name(name: Optional[QueryParam]) -> APIResponse:
_name = name or "Molten"
return APIResponse(
status=200, message=f"Hello, {_name}! Glad you are programming with us"
)
def ping() -> APIResponse:
return APIResponse(200, "Pong")
@annotate(openapi_tags=["Stores"])
def get_stores(store_manager: StoreManager) -> List[Store]:
"""Returns a collection of Stores"""
stores = store_manager.get_stores()
return stores
@annotate(openapi_tags=["Stores"])
def create_store(
store: Store, store_manager: StoreManager
) -> Tuple[Any, Store, Dict[str, str]]:
"""Creates a new store resource and returns its respresentation"""
try:
_store = store_manager.create_store(store)
except HTTPError as err:
raise err
headers = {"Location": _store.href}
return HTTP_201, _store, headers
@annotate(openapi_tags=["Stores"])
def get_store_by_id(id: int, store_manager: StoreManager) -> Store:
_store = store_manager.get_store_by_id(id)
if _store is None:
raise HTTPError(
HTTP_404,
{
"status": 404,
"message": f"The resource you are looking for /stores/{id} does not exist",
},
)
return _store
@annotate(openapi_tags=["Stores"])
def update_store(store: Store, store_manager: StoreManager) -> Store:
_updated_store = store_manager.update_store(store)
return _updated_store
@annotate(openapi_tags=["Stores"])
def delete_store_by_id(id: int, store_manager: StoreManager) -> Tuple[Any, APIResponse]:
store_manager.delete_store_by_id(id)
return (
HTTP_202,
APIResponse(status=202, message=f"Delete request for store: {id} accepted"),
)
@annotate(openapi_tags=["Geographies"])
def get_geographies(geo_manager: GeographyManager) -> List[Geography]:
_geographies = geo_manager.get_geographies()
return _geographies
@annotate(openapi_tags=["Geographies"])
def get_geo_by_id(id: int, geo_manager: GeographyManager) -> Geography:
try:
_geo = geo_manager.get_geo_by_id(id=id)
except HTTPError as err:
raise err
return _geo
@annotate(openapi_tags=["Geographies"])
def get_geo_stores(id: int, store_manager: StoreManager) -> List[Store]:
stores = store_manager.get_stores_by_geo(geo_id=id)
return stores
@annotate(openapi_tags=["Alerts"])
def get_alerts(alert_manager: AlertManager) -> List[Alert]:
alerts = alert_manager.get_alerts()
return alerts
@annotate(openapi_tags=["Alerts"])
def get_alert_by_id(id: int, alert_manager: AlertManager) -> Alert:
try:
alert = alert_manager.get_alert_by_id(alert_id=id)
except EntityNotFound as err:
raise HTTPError(HTTP_404, APIResponse(status=404, message=err.message))
return alert
@annotate(openapi_tags=["Alerts"])
def get_store_alerts(id: int, alert_manager: AlertManager) -> List[Alert]:
try:
alerts = alert_manager.get_alerts_at_store(store_id=id)
except EntityNotFound as err:
raise HTTPError(HTTP_404, APIResponse(status=404, message=err.message))
return alerts
@annotate(openapi_tags=["Alerts"])
def get_geo_alerts(id: int, alert_manager: AlertManager) -> List[Alert]:
try:
alerts = alert_manager.get_alerts_at_geo(geo_id=id)
except EntityNotFound as err:
raise HTTPError(HTTP_404, APIResponse(status=404, message=err.message))
return alerts
routes = [
Route("/", method="GET", handler=name),
Route("/ping", method="GET", handler=ping),
# Stores
Route("/stores", method="GET", handler=get_stores),
Route("/stores", method="POST", handler=create_store),
Route("/stores/{id}", method="GET", handler=get_store_by_id),
Route("/stores/{id}", method="PUT", handler=update_store),
Route("/stores/{id}", method="DELETE", handler=delete_store_by_id),
Route("/geographies/{id}/stores", method="GET", handler=get_geo_stores),
# Geographies
Route("/geographies", method="GET", handler=get_geographies),
Route("/geographies/{id}", method="GET", handler=get_geo_by_id),
# Alerts
Route("/alerts", method="GET", handler=get_alerts),
Route("/alerts/{id}", method="GET", handler=get_alert_by_id),
Route("/stores/{id}/alerts", method="GET", handler=get_store_alerts),
Route("/geographies/{id}/alerts", method="GET", handler=get_geo_alerts),
# OpenAPI
Route("/_schema", get_schema),
Route("/docs", get_docs),
]
components = [
SettingsComponent(settings),
SQLAlchemyEngineComponent(),
SQLAlchemySessionComponent(),
StoreManagerComponent(),
GeographyManagerComponent(),
AlertManagerComponent()
]
middleware = [ResponseRendererMiddleware(), SQLAlchemyMiddleware()]
renderers = [ExtJSONRender()]
app = MarketLook(
routes=routes, components=components, middleware=middleware, renderers=renderers
)

87
market_look/cli.py
Unescape Escape View File

@ -0,0 +1,87 @@
from decimal import Decimal
import click
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from werkzeug.serving import run_simple
from .app import app, settings
from .model import Base, UserModel, GeographyModel, StoreModel, AlertModel
engine = create_engine(settings.get("database_engine_dsn"))
Session = sessionmaker(bind=engine)
@click.group()
def cli():
pass
@click.command()
def serve(host="0.0.0.0", port=8080, debug=True, use_reloader=True):
run_simple(host, port, app, use_debugger=debug, use_reloader=use_reloader)
@click.command()
def initdb():
click.echo("Creating database")
Base.metadata.create_all(bind=engine)
click.echo("Database created")
@click.command()
def dropdb():
click.echo("Are you sure you would like to drop the database?: [Y/N]")
response = input()
if response.lower() == "y":
Base.metadata.drop_all(bind=engine)
click.echo("Database dropped")
else:
click.echo("Database drop aborted")
@click.command()
@click.argument("email", type=str)
@click.argument("passwd", type=str)
def adduser(email, passwd):
user = UserModel(email=email, password=passwd, admin=False)
session = Session()
session.add(user)
session.commit()
click.echo(f"New user {user.email} has been created")
session.close()
@click.command()
def testdata():
session = Session()
user = UserModel(email="test@kellogg.com", password="Welcome1")
geography = GeographyModel(id=500_000, name="TEST", user=user)
store = StoreModel(
id=9_000_000_000,
name="SUPER TEST STORE 4000",
number="4000",
address="123 FAKE STREET",
city="PORTAGE",
zip="49024",
state="MI",
lat=Decimal("42.2607409"),
long=Decimal("-85.6121787"),
tdlinx="0123456",
geography=geography,
)
alert = AlertModel(store=store, promo_name="BATMAN CHEEZIT")
session.add_all([user, geography, store, alert])
session.commit()
click.echo("Test Data created")
session.close()
cli.add_command(serve)
cli.add_command(initdb)
cli.add_command(dropdb)
cli.add_command(adduser)
cli.add_command(testdata)
if __name__ == "__main__":
cli()

9
market_look/errors.py
Unescape Escape View File

@ -0,0 +1,9 @@
from molten.errors import MoltenError
class MarketLookError(MoltenError):
"""Base class for MarketLook errors"""
class EntityNotFound(MarketLookError):
"""Raised when an entity is not found using an `exists` check in sqlalchemy."""

261
market_look/manager.py
Unescape Escape View File

@ -0,0 +1,261 @@
from abc import ABCMeta, abstractmethod
from inspect import Parameter
import typing
from sqlalchemy.orm import Session
from molten import BaseApp, HTTPError, HTTP_409, HTTP_404
from .errors import EntityNotFound
from .model import StoreModel, GeographyModel, AlertModel
from .schema import Alert, Geography, Link, Store, APIResponse
class BaseManager(metaclass=ABCMeta):
"""Base instance for Model managers"""
def __init__(self, session: Session, app: BaseApp):
self.session = session
self.app = app
@abstractmethod
def _model_from_schema(self, schema):
"""Converts a Schema instance into a SQLAlchemy ORM model instance"""
pass
@abstractmethod
def _schema_from_model(self, result):
"""Converts a SQLAlchemy results proxy into a Schema instance"""
pass
class StoreManager(BaseManager):
"""A `StoreManager` is accountable for the CRUD operations associated with a Store"""
def _schema_from_model(self, result: StoreModel) -> Store:
_store = Store(
id=result.id,
href=self.app.reverse_uri("get_store_by_id", id=result.id),
createdDate=result.created_date,
modifiedDate=result.modified_date,
name=result.name,
number=result.number,
address=result.address,
city=result.city,
state=result.state,
zip=result.zip,
lat=result.lat,
long=result.long,
tdlinx=result.tdlinx,
geography=Link(
href=self.app.reverse_uri("get_geo_by_id", id=result.geography_id)
)
if result.geography_id
else None,
alerts=Link(
href=self.app.reverse_uri("get_store_alerts", id=result.id)
)
)
return _store
def _model_from_schema(self, store: Store) -> StoreModel:
_store_model = StoreModel(
id=store.id,
name=store.name,
number=store.number,
address=store.address,
city=store.city,
state=store.state,
zip=store.zip,
lat=store.lat,
long=store.long,
tdlinx=store.tdlinx,
)
return _store_model
def get_stores(self) -> typing.List[Store]:
"""Retrieves a list of Store representations"""
results = self.session.query(StoreModel).order_by(StoreModel.id).all()
_stores = [self._schema_from_model(result) for result in results]
return _stores
def get_store_by_id(self, id: int) -> Store:
"""Retrieves a store representation by id"""
result = self.session.query(StoreModel).filter_by(id=id).one_or_none()
if result is None:
return result
_store = self._schema_from_model(result)
return _store
def create_store(self, store: Store) -> Store:
"""Creates a new store resource and returns its representation"""
result = self.session.query(StoreModel).filter_by(id=store.id).one_or_none()
if result is not None:
raise HTTPError(
HTTP_409,
{
"status": 409,
"message": f"A store with id: {store.id} already exists",
},
)
store_model = self._model_from_schema(store)
self.session.add(store_model)
self.session.flush()
_store = self._schema_from_model(store_model)
return _store
def update_store(self, store: Store) -> Store:
result = self.session.query(StoreModel).filter_by(id=store.id).one_or_none()
_updates = self._model_from_schema(store)
self.session.merge(_updates)
self.session.flush()
_store = self._schema_from_model(result)
return _store
def delete_store_by_id(self, id):
_store = self.session.query(StoreModel).filter_by(id=id).one_or_none()
if _store is not None:
self.session.delete(_store)
return
def get_stores_by_geo(self, geo_id) -> typing.List[Store]:
results = self.session.query(StoreModel).filter_by(geography_id=geo_id).all()
_stores = [self._schema_from_model(store) for store in results]
return _stores
class GeographyManager(BaseManager):
def _model_from_schema(self, schema: Geography) -> GeographyModel:
pass
def _schema_from_model(self, result: GeographyModel) -> Geography:
_geography = Geography(
id=result.id,
href=self.app.reverse_uri("get_geo_by_id", id=result.id),
createdDate=result.created_date,
modifiedDate=result.modified_date,
name=result.name,
stores=Link(href=self.app.reverse_uri('get_geo_stores', id=result.id)),
alerts=Link(href=self.app.reverse_uri('get_geo_alerts', id=result.id)),
)
return _geography
def get_geographies(self) -> typing.List[Geography]:
results = self.session.query(GeographyModel).all()
return [self._schema_from_model(geo) for geo in results]
def get_geo_by_id(self, id: int) -> Geography:
result = self.session.query(GeographyModel).one_or_none()
if result is None:
raise HTTPError(
HTTP_404,
APIResponse(
status=404,
message=f"The resource you are looking for /geographies/{id} does not exist",
),
)
return self._schema_from_model(result)
class AlertManager(BaseManager):
def _model_from_schema(self, schema: Alert) -> AlertModel:
_alert_model = AlertModel(
promo_name=schema.promoName,
response=schema.response,
valid=schema.valid
)
def _schema_from_model(self, result: AlertModel) -> Alert:
_alert = Alert(
id=result.id,
href=self.app.reverse_uri('get_alert_by_id', id=result.id),
createdDate=result.created_date,
modifiedDate=result.modified_date,
promoName=result.promo_name,
response=result.response,
valid=result.valid,
store=Link(href=self.app.reverse_uri('get_store_by_id', id=result.store.id))
)
return _alert
def get_alerts(self) -> typing.List[Alert]:
results = self.session.query(AlertModel).all()
alerts = [self._schema_from_model(alert) for alert in results]
return alerts
def get_alert_by_id(self, alert_id) -> Alert:
result = self.session.query(AlertModel).filter_by(id=alert_id).one_or_none()
if result is None:
raise EntityNotFound(f"Alert: {alert_id} does not exist")
alert = self._schema_from_model(result)
return alert
def get_alerts_at_store(self, store_id) -> typing.List[Alert]:
store_check = self.session.query(StoreModel.id).filter_by(id=store_id).exists()
if not self.session.query(store_check).scalar():
raise EntityNotFound(f'Store {store_id} does not exist')
results = self.session.query(AlertModel).filter_by(store_id=store_id, active=True).all()
alerts = [self._schema_from_model(alert) for alert in results]
return alerts
def get_alerts_at_geo(self, geo_id) -> typing.List[Alert]:
geo_check = self.session.query(GeographyModel.id).filter_by(id=geo_id).exists()
if not self.session.query(geo_check).scalar():
raise EntityNotFound(f'Geography {geo_id} does not exist')
_subquery = self.session.query(StoreModel.id).filter(StoreModel.geography_id==geo_id)
results = self.session.query(AlertModel).filter(AlertModel.store_id.in_(_subquery)).all()
alerts = [self._schema_from_model(alert) for alert in results]
return alerts
def create_alert_at_store(self, store_id, alert: Alert) -> Alert:
store = self.session.query(StoreModel).filter_by(id=store_id).one()
alert_model = self._model_from_schema(alert)
alert_model.store = store
self.session.add(alert_model)
self.session.flush()
_alert = self._schema_from_model(alert_model)
return _alert
class StoreManagerComponent:
is_cacheable = True
is_singleton = False
def can_handle_parameter(self, parameter: Parameter) -> bool:
return parameter.annotation is StoreManager
def resolve(self, session: Session, app: BaseApp) -> StoreManager: # type: ignore
return StoreManager(session, app)
class GeographyManagerComponent:
is_cacheable = True
is_singleton = False
def can_handle_parameter(self, parameter: Parameter) -> bool:
return parameter.annotation is GeographyManager
def resolve(self, session: Session, app: BaseApp) -> StoreManager: # type: ignore
return GeographyManager(session, app)
class AlertManagerComponent:
is_cacheable = True
is_singleton = False
def can_handle_parameter(self, parameter: Parameter) -> bool:
return parameter.annotation is AlertManager
def resolve(self, session: Session, app: BaseApp) -> AlertManager: # type: ignore
return AlertManager(session, app)

103
market_look/model.py
Unescape Escape View File

@ -0,0 +1,103 @@
import bcrypt
from sqlalchemy import (
Column,
Integer,
BigInteger,
Numeric,
String,
ForeignKey,
DateTime,
Boolean,
Enum,
)
from sqlalchemy.orm import relationship
from sqlalchemy.sql import expression
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.types import DateTime as DatetimeType
BCRYPT_LOG_ROUNDS = 11
Base = declarative_base()
class utcnow(expression.FunctionElement):
type = DatetimeType()
@compiles(utcnow, "postgresql")
def pg_utcnow(element, compiler, **kw):
return "TIMEZONE('utc', CURRENT_TIMESTAMP)"
class DBMixin:
id = Column(BigInteger, primary_key=True)
created_date = Column(DateTime(timezone=True), server_default=utcnow())
modified_date = Column(
DateTime(timezone=True), server_default=utcnow(), onupdate=utcnow()
)
class UserModel(DBMixin, Base):
__tablename__ = "user"
email = Column(String(255), nullable=False, unique=True)
password = Column(String(255))
geography = relationship("GeographyModel", back_populates="user")
admin = Column(Boolean, nullable=False, default=False)
confirmed = Column(Boolean, nullable=False, default=False)
active = Column(Boolean, nullable=False, default=True)
def __init__(self, email, password, admin=False):
self.email = email
self.password = bcrypt.hashpw(
password.encode("utf-8"), bcrypt.gensalt(BCRYPT_LOG_ROUNDS)
).decode()
self.admin = admin
def check_password(self, password):
return bcrypt.checkpw(password.encode("utf-8"), self.password.encode("utf-8"))
class GeographyModel(Base, DBMixin):
__tablename__ = "geography"
name = Column(String(255))
user_id = Column(BigInteger, ForeignKey("user.id"), nullable=True)
active = Column(Boolean, nullable=False, default=True)
user = relationship("UserModel", uselist=False, back_populates="geography")
stores = relationship(
"StoreModel", order_by="StoreModel.id", back_populates="geography"
)
class StoreModel(Base, DBMixin):
__tablename__ = "store"
geography_id = Column(BigInteger, ForeignKey("geography.id"), nullable=True)
name = Column(String(255))
number = Column(String(4), nullable=True)
address = Column(String(255))
city = Column(String(255))
state = Column(String(2))
zip = Column(String(12))
lat = Column(Numeric(10, 7), nullable=True)
long = Column(Numeric(10, 7), nullable=True)
tdlinx = Column(String(8), nullable=True)
active = Column(Boolean, nullable=False, default=True)
geography = relationship("GeographyModel", back_populates="stores")
alerts = relationship("AlertModel", back_populates="store")
class AlertModel(Base, DBMixin):
__tablename__ = "alert"
store_id = Column(BigInteger, ForeignKey("store.id"))
store = relationship("StoreModel", back_populates="alerts")
promo_name = Column(String(255))
response = Column(
Enum("", "manager_refused", "valid", "invalid", name="response_type")
)
responded = Column(Boolean, default=False)
valid = Column(Boolean, nullable=True)
active = Column(Boolean, nullable=False, default=True)

79
market_look/schema.py
Unescape Escape View File

@ -0,0 +1,79 @@
from typing import Optional, List, Union
from molten import schema, field
@schema
class Link:
href: str
@schema
class Geography:
id: int
href: str = field(response_only=True)
createdDate: str = field(response_only=True)
modifiedDate: str = field(response_only=True)
name: str
stores: Link = field(response_only=True)
alerts: Link = field(response_only=True)
@schema
class Store:
id: int
href: str = field(response_only=True)
createdDate: str = field(response_only=True)
modifiedDate: str = field(response_only=True)
name: str
number: Optional[str]
address: str
city: str
state: str
zip: str
lat: Optional[float]
long: Optional[float]
tdlinx: Optional[str]
geography: Link = field(response_only=True)
alerts: Link = field(response_only=True)
@schema
class User:
id: int = field(response_only=True)
href: str = field(response_only=True)
email: str
password: str = field(request_only=True)
geography: Link = field(response_only=True)
@schema
class Alert:
id: int
href: str = field(response_only=True)
createdDate: str = field(response_only=True)
modifiedDate: str = field(response_only=True)
promoName: str
response: Optional[str]
valid: Optional[bool]
store: Link = field(response_only=True)
@schema
class StorePatch:
name: Optional[str]
number: Optional[str]
address: Optional[str]
city: Optional[str]
state: Optional[str]
zip: Optional[str]
lat: Optional[float]
long: Optional[float]
tdlinx: Optional[str]
@schema
class APIResponse:
status: int = field(description="An HTTP status code")
message: str = field(
description="A user presentable message in response to the request provided to the API"
)

21
market_look/util.py
Unescape Escape View File

@ -0,0 +1,21 @@
import datetime as dt
from decimal import Decimal
from typing import Any
from molten import JSONRenderer, is_schema, dump_schema
class ExtJSONRender(JSONRenderer):
"""JSON Render with support for ISO 8601 datetime format strings"""
def default(self, ob: Any) -> Any:
"""You may override this when subclassing the JSON renderer in
order to encode non-standard object types.
"""
if is_schema(type(ob)):
return dump_schema(ob)
if isinstance(ob, dt.datetime):
return ob.isoformat()
if isinstance(ob, Decimal):
return float(ob)
raise TypeError(f"cannot encode values of type {type(ob)}") # pragma: no cover

5
requirements.in
Unescape Escape View File

@ -0,0 +1,5 @@
bcrypt
molten
psycopg2
sqlalchemy

19
requirements.txt
Unescape Escape View File

@ -0,0 +1,19 @@
#
# This file is autogenerated by pip-compile
# To update, run:
#
# pip-compile --output-file requirements.txt requirements.in
#
--trusted-host pypi.python.org
--trusted-host git.aigalatic.com
bcrypt==3.1.4
cffi==1.11.5 # via bcrypt
molten==0.5.0
mypy-extensions==0.4.0 # via typing-inspect
psycopg2==2.7.5
pycparser==2.18 # via cffi
six==1.11.0 # via bcrypt
sqlalchemy==1.2.10
typing-extensions==3.6.5 # via molten
typing-inspect==0.3.1 # via molten

15
setup.py
Unescape Escape View File

@ -0,0 +1,15 @@
from setuptools import setup, find_packages
setup(
name="Market Look",
description="An API application used for tracking promotion compliance at store",
zip_safe=False,
author="Drew Bednar",
author_email="andrew.bednar@kellogg.com",
packages=find_packages(exclude=['tests'], include=['scripts', 'market_look']),
entry_points={
"console_scripts": [
"manage=market_look.cli:cli"
]
}
)
Write Preview
Loading…
Cancel
Save