Passwd

[Python] sqlite3 - SQLite 사용하기

개요이전에 2023.01.08-[SQLite] 설치 및 사용에서 간단한 파일 데이터베이스인 SQLite를 설치하고 실행해 보았다. 당시에는 CLI를 활용하여 SQLite를 다뤘는데, 이번에는 Python으로 SQLite를 다뤄보려고 한다.방법을 알아보자. sqlite3SQLite 데이터베이스용 라이브러리인 sqlite3는 Python 기본 내장 라이브러리로 별도로 설치하지 않고 바로 import 하여 사용할 수 있다.import sqlite3 데이터베이스 연결connect 함수를 사용하여 데이터베이스 연결을 생성한다. 기본적으로 connect 함수로 전달한 값이 파일 DB의 경로가 된다.import sqlite3con = sqlite3.connect('example.db')아무 테이블도 생성하지 않..

개요Jupyter 노트북 환경에서 작업을 하다 보면 DataFrame을 확인할 때 컬럼이나 행이 생략되어 전체 정보를 확인할 수 없는 경우가 있다. 이런 경우 생략하는 정도를 조정하면 보다 효율적으로 데이터를 조회할 수 있다. 현재 옵션 확인옵션을 조정하기 전에 먼저 어떤 옵션이 있는지 확인할 필요가 있을 것 같다.옵션에 대한 설명은 pandas.describe_option을 통해 확인할 수 있다.import pandas as pdpd.describe_option()코드 상으로도 확인할 수 있지만 공식 문서를 활용하는 게 더 수월할 수 있다.https://pandas.pydata.org/pandas-docs/stable/user_guide/options.html#available-options 인쇄..

개요Python으로 비동기 프로그래밍을 진행 중인데, 프로그램 결과 파일을 생성해야 한다. 그런데 Python에서 파일을 다룰 때 사용하는 open 함수는 비동기를 지원하지 않는다고 한다. 이 문제를 해결하기 위해 Python 비동기 프로그래밍으로 파일 작업을 할 수 있도록 지원해 주는 ailfiles라는 라이브러리 사용 방법을 알아본다.   aiofilesaiofiles은 비동기 애플리케이션에서 로컬 디스크의 파일을 처리하기 위해 Python으로 작성한 Apache 2 라이센스의 라이브러리다.일반적으로 로컬 파일 IO는 블로킹이기 때문에 쉽게 비동기 방식으로 수행할 수 없다. 다시 말해 파일 IO 수행은 비동기 애플리케이션에 영향을 줄 수 있기 때문에, 실행 중인 스레드를 차단해서는 안 된다. 하지만 ..

개요2025.04.06-[Python] websockets - 연결 종료 시 자동 재연결에서 연결이 자동으로 종료되는 경우, 자동으로 재연결을 시도하도록 코드를 개선했었다.확인해보니 어떤 웹소켓 서버의 경우에는 기본적으로 아무 데이터도 수/발신하지 않는 상태로 120초 정도가 경과하면 웹소켓 연결을 종료한다고 한다. 때문에 websockets 라이브러리는 클라이언트에서 서버와의 연결을 확인함과 동시에 keepalive 역할을 할 수 있는 ping 함수를 제공하고 있다. 사용 방법을 가볍게 알아본다.  websockets.pingwebsockets.ping 함수는 0x9 연산코드를 포함하는 Ping 프레임을 보낸다. pong_waiter = await websocket.ping()# only if you ..

개요Python으로 WebSocket 통신이 필요한 프로그램을 개발하고 있는데, 여러 웹소켓 통신 연결 및 유지가 필요한 상황이다. 평소 Python으로 개발을 할 때 특정 서버, 그것도 여러 서버와 지속적인 통신 연결을 유지할 상황이 많지 않았다. 이번 기회에 적당한 방법을 정리해 둔다.   웹소켓 통신 함수먼저 웹소켓 통신을 유지할 함수를 작성한다. 함수는 2025.04.06-[Python] websockets - 연결 종료 시 자동 재연결에서 작성한 코드이다. 다만 재사용성을 확보할 수 있도록 통신할 URL과 전송할 데이터를 매개변수로 전달받을 수 있도록 수정했다.import asynciofrom websockets.asyncio.client import connectimport jsonasync ..

개요Python websockets 라이브러리르 사용해서 가상 자산 거래소에 orderbook 정보를 구독하고 있는데, 일정 시간이나 횟수가 지나면 자동으로 연결이 종료되는 문제가 발생했다.관련해서 자동으로 연결을 재시도할 수 있도록 코드를 변경해보려고 한다.   기존 코드import asynciofrom websockets.asyncio.client import connectimport jsonasync def connect_test(): url = "wss://ws-api.korbit.co.kr/v2/public" send_data = [ { "method": "subscribe", "type": "orderbook", ..

개요최근 WebSocket 통신을 다룰 일이 생겼다. 관련하여 Websocket 서버와 클라이언트를 구축하기 위한 Python 라이브러리인 websockets 사용 예시를 간단하게 적어두려고 한다.websockets 라이브러리는 기본적으로 Python 내장 비동기 I/O 라이브러리인 asyncio를 기반으로 한다. threading 구현을 제공하고 있어 asyncio에 익숙하지 않은 경우의 대안을 제공하기도 한다.  WebSocket이란먼저 WebSocket 개념에 대해 빠르게 정리한다.WebSocket은 클라이언트와 서버 간의 양방향 데이터 전송을 가능하게 하는 실시간 통신 프로토콜이다. 비상태 프로토콜인 HTTP와 달리 클라이언트와 서버 간의 지속적인 연결을 유지하여 즉각적인 데이터 전송을 허용한다...

개요Pandas를 사용하면 주로 딕셔너리, 배열 등을 사용해 데이터프레임을 만들거나 파일을 읽어서, 데이터 처리를 수행하게 된다. 하지만 단순한 테스트 경우에는 간단히 문자열 데이터를 사용해 데이터프레임을 만들게 되는데, 이번 글에서는 문자열로만 데이터프레임을 만드는 방법을 소개하려고 한다.   방법Pandas에서 파일을 읽어 들여 데이터프레임을 만드는 함수는 대개 매개변수로 파일 경로나 버퍼를 전달받는다. 즉, read__csv나 read_json 같은 함수에 파일 경로가 아니라 문자열 버퍼를 전달하면 된다. 방법은 무척 간단하다. 문자열 값을 io.StringIO를 사용해 StringIO 객체로 만들어 read_csv 함수 등에 전달하면 된다.참고로 StringIO는 인메모리 텍스트 버퍼를 사용하는 ..

개요알고리즘 문제를 풀다 보면 문자열이나 리스트에서 연속적인 부분을 찾아야 하는 문제를 찾아볼 수 있다. 직접 순회하면서 확인하는 방법도 있지만, 이번 글에서는 itertools의 groupby를 사용해 찾아보려고 한다.  iterrtools.groupbygroupby는 연속적인 키과 그룹을 반환하는 iterator를 만든다. 키는 각 요소에 대한 키 값을 계산하는 함수로, 지정하지 않으면 항등 함수로 기본 설정된다. 키 값이 변경될 때마다 그룹이 새로 생성되기 때문에 기본적으로 정렬된 상태에서 사용해야 한다.  사용 예시예로 들어 abbbaaaa라는 문자열에서 연속적인 부분을 찾으려고 한다.from itertools import groupbystr_input = 'abbbaaaa'for k, val i..

개요운영 중인 서비스에서 Trino를 백엔드로 사용하고 있는데, 기존에는 Trino CLI나 별도의 프런트엔드를 사용해서 작업을 수행했었다. 최근 프로그래밍적인 방법을 사용해서 작업을 할 필요가 있어, 방법을 적어두려고 한다.사용 언어는 Python이다.  필요 라이브러리 설치Python으로 Trino를 다루기 위해서는 trino 클라이언트 라이브러리를 선택할 수 있다. trino 클라이언트 라이브러리는 아래와 같이 pip 명령어로 설치할 수 있는데…… pip install trino테스트 환경이 컨테이너 기반이라서 그런지 trino 라이브러리로는 연결이 잘 안 됐다. 때문에 이 글에서는 sqlalchemy를 사용해 다뤄보려고 한다.pip install sqlalchemy  사용법연결 URLsqlalch..