victorusachev

from fabric.api import local
import pip

def freeze ():
    local("pip freeze > requirements.txt")
    local("git add requirements.txt")
    local("git commit -v")

def upgrade ():
    for dist in pip.get_installed_distributions():

victorusachev

# The contents of this file are subject to the Common Public Attribution
# License Version 1.0. (the "License"); you may not use this file except in
# compliance with the License. You may obtain a copy of the License at
# http://code.reddit.com/LICENSE. The License is based on the Mozilla Public
# License Version 1.1, but Sections 14 and 15 have been added to cover use of
# software over a computer network and provide for limited attribution for the
# Original Developer. In addition, Exhibit A has been modified to be consistent
# with Exhibit B.
#
# Software distributed under the License is distributed on an "AS IS" basis,

victorusachev

#!/usr/bin/env python
# –*– encoding: UTF-8 –*–

'''ftp_updown.py: connection to FTP and up & down function'''

__author__ 		= 'frantisekrehor.cz'
__email__		= '[email protected]'

#================================================================
	

victorusachev

def update_remains_period(actor, start, end, company_object):
    """
    Обновить остатки топлива за период.
    """
    fuel_infos = FuelInfo.objects.filter(active=True, boiler=company_object)
    for date in get_period_dates(start, end):
        for fuel_info in fuel_infos:
            # Запись, которую нужно обновить
            record = None
            try:

victorusachev

from pprint import pprint

from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, ForeignKey, Sequence, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import backref, relationship, sessionmaker


engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)

victorusachev

from collections import namedtuple


CreditCondition = namedtuple(
    'CreditCondition',
    ['property_value',
     'initial_payment',
     'loan_months',
     'interest_rate'])

victorusachev

import datetime
import functools
from typing import Callable, Optional


def timed_cache(timedelta: datetime.timedelta,
                maxsize: Optional[int] = 128,
                typed: Optional[bool] = False) -> Callable:
    """Timeout is applied to the whole cache."""
    def decorator(func):

victorusachev

class Predicate:
    def __init__(self, pattern):
        self.pattern = pattern

    def __call__(self, value):
        return self.pattern in value

    def __repr__(self):
        return f'<Predicate: {self.pattern}>'

victorusachev

import asyncio
import time


async def do_some_prints_with_sleeps(coro_id):
    print(f"[{coro_id}] Hello, i am coro with asyncio.sleeps")
    print(f"[{coro_id}] Will sleep right now for 1 second")
    await asyncio.sleep(1)
    print(f"[{coro_id}] I woke up, but feel sleepy, will sleep 1 second more")
    await asyncio.sleep(1)

victorusachev

# convert to dict
# fiasHouseGuidByAddress = addressObjects
#     .Select(
#         ao => new KeyValuePair<string, Guid>(ao.AddressObjectId, ao.FiasHouseGuid)
#     )
#     .ToDictionary(t => t.Key, t => t.Value);

import uuid
from dataclasses import dataclass, field
from pprint import pprint