semvec/pypi-packages huggingface dataset:
ColumnAlchemy 0.10.1
additional column types for SQLAlchemy
License: MIT
Requires Python: >=3.6


This repo contains a **custom column types for SQLAlchemy** which I use in different projects.

The code should be **compatible with SQLAlchemy 1.2.3 - 2.0** and **Python 3.6 - 3.13** as well as **pypy 3.9**.

    pip install ColumnAlchemy


UTCDateTime
-----------

`UTCDateTime` stores a Python tz-aware `datetime.datetime` value as UTC datetime in the database (without explicit timezone information). I use this to introduce tz-aware timezones in systems which expect "naive" datetimes in the database.

    from schwarz.column_alchemy import UTCDateTime

    class Foo(Base)
        __tablename__ = 'foo'
        id = Column(Integer, autoincrement=True, primary_key=True)
        timestamp = Column(UTCDateTime)


ShiftedDecimal
--------------

`ShiftedDecimal` stores a `Decimal` as integer in the database (with limited precision). This is especially useful to store decimal values even in sqlite which requires special treatment to store decimals.

    from decimal import Decimal
    from schwarz.column_alchemy import ShiftedDecimal

    class Foo(Base)
        __tablename__ = 'foo'
        id = Column(Integer, autoincrement=True, primary_key=True)
        percentage = Column(ShiftedDecimal(4))
    
    foo = Foo(percentage=Decimal('1.2324'))
    # stores percentage as 12324 in the database but returns the
    # correct Decimal value after loading.


ValuesEnum
-----------
TODO

IntValuesEnum
-------------
TODO


YearMonthColumn
---------------

`YearMonth` is similar to `datetime.date` but without a `day` attribute. It can be used to represent a calendar month and provides some convenience methods like `first_date_of_month()` and `last_date_of_month()`. A **`YearMonthColumn`** stores a `YearMonth` instance as "YYYY-MM" in the database.

    from schwarz.column_alchemy import YearMonth, YearMonthColumn

    class Foo(Base)
        __tablename__ = 'foo'
        id = Column(Integer, autoincrement=True, primary_key=True)
        month = Column(YearMonthColumn())

    foo = Foo(month=YearMonth(2020, 7))
    # stores "month" as "2020-07" in the database but returns a
    # YearMonth instance after loading.


YearMonthIntColumn
------------------

Very similar to `YearMonthColumn` but stores `YearMonth(2020, 7)` as `202007` (integer).
rabbitmq-clients 0.1.22
High level interfaces for work with RabbitMQ.
License: MIT
Requires Python: <4.0,>=3.10

[![PyPI version](https://badge.fury.io/py/rabbitmq-clients.svg)](https://pypi.org/project/rabbitmq-clients/)
[![GitHub Build](https://github.com/Deskent/rabbit_mq_client/workflows/Tests/badge.svg)](https://github.com/Deskent/rabbit_mq_client/actions)
[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/rabbitmq-clients.svg)](https://pypi.org/project/rabbitmq-clients/)


### Stack:

- [x] <a href="https://www.python.org/"><img src="https://raw.githubusercontent.com/devicons/devicon/master/icons/python/python-plain.svg" alt="python" width="15" height="15"/>
  Python 3.10+ <br/></a>
- [x] <a href="https://github.com/mosquito/aio-pika"><img src="https://raw.githubusercontent.com/mosquito/aio-pika/b61062893c4973dbbd5ac6a6afa55e4e74b28ae5/logo.svg" alt="aio-pika" width="15" height="15"/>
  aio-pika 9.4.1+ <br/></a>


### Installation

    pip install rabbitmq-clients

### Usage

    Run rabbitMQ on your computer.

#### Producer

    import asyncio
    import json

    from rabbitmq_clients import RabbitProducer

    if __name__ == "__main__":
        producer = RabbitProducer(
            host='localhost',
            login='test',
            password='test',
        )
        data = {"result": True}
        asyncio.run(
            producer.publish(
                json.dumps(data),
                queue_name='test_queue_name',
            )
        )


#### Consume multiply queues (recommended)

    import asyncio

    from rabbitmq_clients import RabbitConsumer, QueueDTO
    from rabbitmq_clients.core.types import JSON


    async def show_decoded_result(result: JSON):
        print(result)


    async def show_encoded_result(result: JSON):
        print(result.body)


    if __name__ == "__main__":
        consumer = RabbitConsumer(
            host='localhost',
            login='test',
            password='test',
        )
        first_queue = QueueDTO(
            name='decoded_result',
            callback=show_decoded_result,
        )
        second_queue = QueueDTO(
            name='encoded_result',
            callback=show_encoded_result,
            auto_decode=False,
        )
        consumer.add_queue(first_queue)
        consumer.add_queue(second_queue)

        try:
            asyncio.run(consumer.consume_all())
        except KeyboardInterrupt:
            pass