This file contains hidden or 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
from sqlalchemy.dialects import postgresql | |
def upsert(session, model, fields, returning=False, return_columns=None): | |
table = model.__table__ | |
stmt = postgresql.insert(table).values(**fields) | |
pk_columns = table.primary_key.columns.keys() | |
update_cols = {col: val for col, val in fields.items() if col not in pk_columns} |
This file contains hidden or 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
FROM python:3.7-alpine3.9 | |
LABEL maintainer='[email protected]' | |
LABEL description='Python services base image' | |
# Don't periodically check PyPI to determine whether a new version of pip is available for download. | |
ENV PIP_DISABLE_PIP_VERSION_CHECK=on | |
# Disable package cache. | |
ENV PIP_NO_CACHE_DIR=off |
This file contains hidden or 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
@pytest.fixture(autouse=True) | |
def mock_request(): | |
with responses.RequestsMock() as req: | |
yield req | |
@pytest.fixture() | |
def check_request(mock_request): | |
def _check_request(body): | |
if isinstance(body, dict): |
This file contains hidden or 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
from datetime import timezone | |
import sqlalchemy as sa | |
class UTCDateTime(sa.TypeDecorator): # pylint:disable=W0223 | |
"""By default if we provide datetime object without timezone Postgres applies | |
his timezone to that datetime. To prevent unexpected behaviour we add utc timezone | |
before every write. And convert back to utc, after every read. |
This file contains hidden or 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
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base, declared_attr | |
from sqlalchemy import MetaData | |
from sqlalchemy.engine import create_engine | |
from sqlalchemy.orm import Session | |
from sqlalchemy.orm import scoped_session as _scoped_session | |
from sqlalchemy.orm import sessionmaker as _sessionmaker | |
from .config import config | |
engine = create_engine( |
This file contains hidden or 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
from contextlib import ContextDecorator | |
from functools import wraps | |
from typing import Any, Callable, Optional, Type | |
class SessionContext(ContextDecorator): | |
""" | |
class Session(SessionContext): | |
scoped_session = ... | |
This file contains hidden or 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
#!/usr/bin/env python | |
"""Wrapper around Flake8 to enable multiprocessing on all operating systems. | |
As of Python 3.8, macOS's default "start method" for multiprocessing is `spawn`. Flake8 | |
requires a "start method" of `fork`, and disables multiprocessing if it detects `spawn` | |
or some other "start method". This script enables the `fork` start method before passing | |
along any command-line arguments to `flake8`. | |
This has never caused me any problems, but note that they disabled this for a reason: | |
Flake8's plugin interface doesn't work with `spawn`, and the maintainer says that `fork` | |
is "pretty broken" on macOS. | |
See: |